Utilize database-backed unread message count in thread.

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationAdapter.java

@@ -397,10 +397,8 @@ public class ConversationAdapter
     return recipient.getId().equals(recipientId);
   }
 
-  void onBindLastSeenViewHolder(StickyHeaderViewHolder viewHolder, int position) {
+  void onBindLastSeenViewHolder(StickyHeaderViewHolder viewHolder, long unreadCount) {
-    int messagePosition = isTypingViewEnabled ? position - 1 : position;
+    viewHolder.setText(viewHolder.itemView.getContext().getResources().getQuantityString(R.plurals.ConversationAdapter_n_unread_messages, (int) unreadCount, (int) unreadCount));
-    int count = messagePosition + 1;
-    viewHolder.setText(viewHolder.itemView.getContext().getResources().getQuantityString(R.plurals.ConversationAdapter_n_unread_messages, count, count));
 
     if (hasWallpaper) {
       viewHolder.setBackgroundRes(R.drawable.wallpaper_bubble_background_18);

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationFragment.java

@@ -203,6 +203,7 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 
+import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
 import kotlin.Unit;
 
 @SuppressLint("StaticFieldLeak")
@@ -216,6 +217,7 @@ public class ConversationFragment extends LoggingFragment implements Multiselect
   private final ActionModeCallback  actionModeCallback     = new ActionModeCallback();
   private final ItemClickListener   selectionClickListener = new ConversationFragmentItemClickListener();
   private final LifecycleDisposable disposables            = new LifecycleDisposable();
+  private final LifecycleDisposable lastSeenDisposable     = new LifecycleDisposable();
 
   private ConversationFragmentListener listener;
 
@@ -225,7 +227,7 @@ public class ConversationFragment extends LoggingFragment implements Multiselect
   private Locale                      locale;
   private FrameLayout                 videoContainer;
   private RecyclerView                list;
-  private RecyclerView.ItemDecoration lastSeenDecoration;
+  private LastSeenHeader              lastSeenDecoration;
   private RecyclerView.ItemDecoration inlineDateDecoration;
   private ViewSwitcher                topLoadMoreView;
   private ViewSwitcher                bottomLoadMoreView;
@@ -258,7 +260,6 @@ public class ConversationFragment extends LoggingFragment implements Multiselect
   private Colorizer                  colorizer;
   private ConversationUpdateTick     conversationUpdateTick;
   private MultiselectItemDecoration  multiselectItemDecoration;
-  private LifecycleDisposable        lifecycleDisposable;
 
   private @Nullable ConversationData conversationData;
   private @Nullable ChatWallpaper    chatWallpaper;
@@ -286,6 +287,7 @@ public class ConversationFragment extends LoggingFragment implements Multiselect
   @Override
   public View onCreateView(@NonNull LayoutInflater inflater, ViewGroup container, Bundle bundle) {
     disposables.bindTo(getViewLifecycleOwner());
+    lastSeenDisposable.bindTo(getViewLifecycleOwner());
 
     final View view = inflater.inflate(R.layout.conversation_fragment, container, false);
     videoContainer = view.findViewById(R.id.video_container);
@@ -354,9 +356,6 @@ public class ConversationFragment extends LoggingFragment implements Multiselect
 
     giphyMp4ProjectionRecycler = initializeGiphyMp4();
 
-    lifecycleDisposable = new LifecycleDisposable();
-    lifecycleDisposable.bindTo(getViewLifecycleOwner());
-
     this.groupViewModel         = new ViewModelProvider(getParentFragment(), new ConversationGroupViewModel.Factory()).get(ConversationGroupViewModel.class);
     this.messageCountsViewModel = new ViewModelProvider(getParentFragment()).get(MessageCountsViewModel.class);
     this.conversationViewModel  = new ViewModelProvider(getParentFragment(), new ConversationViewModel.Factory()).get(ConversationViewModel.class);
@@ -438,6 +437,9 @@ public class ConversationFragment extends LoggingFragment implements Multiselect
 
     conversationViewModel.getSearchQuery().observe(getViewLifecycleOwner(), this::onSearchQueryUpdated);
 
+    disposables.add(conversationViewModel.getMarkReadRequests()
+                                         .subscribe(timeSince -> markReadHelper.onViewsRevealed(timeSince)));
+
     return view;
   }
 
@@ -975,12 +977,23 @@ public class ConversationFragment extends LoggingFragment implements Multiselect
   }
 
   public void setLastSeen(long lastSeen) {
+    lastSeenDisposable.clear();
     if (lastSeenDecoration != null) {
       list.removeItemDecoration(lastSeenDecoration);
     }
 
     lastSeenDecoration = new LastSeenHeader(getListAdapter(), lastSeen);
     list.addItemDecoration(lastSeenDecoration, 0);
+
+    if (lastSeen > 0) {
+      lastSeenDisposable.add(conversationViewModel.getThreadUnreadCount()
+                                                  .distinctUntilChanged()
+                                                  .observeOn(AndroidSchedulers.mainThread())
+                                                  .subscribe(unreadCount -> {
+                                                    lastSeenDecoration.setUnreadCount(unreadCount);
+                                                    list.invalidateItemDecorations();
+                                                  }));
+    }
   }
 
   private void handleCopyMessage(final Set<MultiselectPart> multiselectParts) {
@@ -1383,7 +1396,7 @@ public class ConversationFragment extends LoggingFragment implements Multiselect
                                           .max(Long::compareTo)
                                           .orElse(0L);
 
-      markReadHelper.onViewsRevealed(Math.max(record.getDateReceived(), latestReactionReceived));
+      conversationViewModel.submitMarkReadRequest(Math.max(record.getDateReceived(), latestReactionReceived));
     }
   }
 
@@ -2126,7 +2139,7 @@ public class ConversationFragment extends LoggingFragment implements Multiselect
   }
 
   private void handleBlockJoinRequest(@NonNull Recipient recipient) {
-    lifecycleDisposable.add(
+    disposables.add(
         groupViewModel.blockJoinRequests(ConversationFragment.this.recipient.get(), recipient)
                       .subscribe(result -> {
                         if (result.isFailure()) {

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationRepository.java

@@ -9,11 +9,14 @@ import androidx.annotation.WorkerThread;
 import org.signal.core.util.concurrent.SignalExecutors;
 import org.signal.core.util.logging.Log;
 import org.thoughtcrime.securesms.contacts.sync.ContactDiscovery;
+import org.thoughtcrime.securesms.database.Database;
+import org.thoughtcrime.securesms.database.DatabaseObserver;
 import org.thoughtcrime.securesms.database.GroupDatabase;
 import org.thoughtcrime.securesms.database.MessageDatabase;
 import org.thoughtcrime.securesms.database.RecipientDatabase;
 import org.thoughtcrime.securesms.database.SignalDatabase;
 import org.thoughtcrime.securesms.database.ThreadDatabase;
+import org.thoughtcrime.securesms.database.model.ThreadRecord;
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
 import org.thoughtcrime.securesms.jobs.MultiDeviceViewedUpdateJob;
 import org.thoughtcrime.securesms.keyvalue.SignalStore;
@@ -162,4 +165,22 @@ class ConversationRepository {
                                           true);
     }).subscribeOn(Schedulers.io());
   }
+
+  Observable<ThreadRecord> getThreadRecord(long threadId) {
+    if (threadId == -1L) {
+      return Observable.empty();
+    }
+
+    return Observable.<ThreadRecord>create(emitter -> {
+
+      DatabaseObserver.Observer listener = () -> {
+        emitter.onNext(SignalDatabase.threads().getThreadRecord(threadId));
+      };
+
+      ApplicationDependencies.getDatabaseObserver().registerConversationObserver(threadId, listener);
+      emitter.setCancellable(() -> ApplicationDependencies.getDatabaseObserver().unregisterObserver(listener));
+
+      listener.onChanged();
+    }).subscribeOn(Schedulers.io());
+  }
 }

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationViewModel.java

@@ -59,6 +59,7 @@ import io.reactivex.rxjava3.core.BackpressureStrategy;
 import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.core.Observable;
 import io.reactivex.rxjava3.disposables.CompositeDisposable;
+import io.reactivex.rxjava3.processors.PublishProcessor;
 import io.reactivex.rxjava3.schedulers.Schedulers;
 import io.reactivex.rxjava3.subjects.BehaviorSubject;
 import kotlin.Unit;
@@ -95,6 +96,8 @@ public class ConversationViewModel extends ViewModel {
   private final RxStore<ConversationState>            conversationStateStore;
   private final CompositeDisposable                   disposables;
   private final BehaviorSubject<Unit>          conversationStateTick;
+  private final RxStore<ThreadCountAggregator> threadCountStore;
+  private final PublishProcessor<Long>         markReadRequestPublisher;
 
   private ConversationIntents.Args args;
   private int                      jumpToPosition;
@@ -123,6 +126,8 @@ public class ConversationViewModel extends ViewModel {
     this.conversationStateStore         = new RxStore<>(ConversationState.create(), Schedulers.io());
     this.disposables                    = new CompositeDisposable();
     this.conversationStateTick          = BehaviorSubject.createDefault(Unit.INSTANCE);
+    this.threadCountStore               = new RxStore<>(ThreadCountAggregator.Init.INSTANCE, Schedulers.computation());
+    this.markReadRequestPublisher       = PublishProcessor.create();
 
     BehaviorSubject<Recipient> recipientCache = BehaviorSubject.create();
 
@@ -132,6 +137,11 @@ public class ConversationViewModel extends ViewModel {
         .map(Recipient::resolved)
         .subscribe(recipientCache);
 
+    disposables.add(threadCountStore.update(
+        threadId.switchMap(conversationRepository::getThreadRecord).toFlowable(BackpressureStrategy.BUFFER),
+        (record, count) -> count.updateWith(record)
+    ));
+
     conversationStateStore.update(Observable.combineLatest(recipientId, conversationStateTick, (id, tick) -> id)
                                             .distinctUntilChanged()
                                             .switchMap(conversationRepository::getSecurityInfo)
@@ -248,6 +258,10 @@ public class ConversationViewModel extends ViewModel {
     }
   }
 
+  void submitMarkReadRequest(long timestampSince) {
+    markReadRequestPublisher.onNext(timestampSince);
+  }
+
   boolean shouldPlayMessageAnimations() {
     return threadAnimationStateStore.getState().shouldPlayMessageAnimations();
   }
@@ -292,6 +306,16 @@ public class ConversationViewModel extends ViewModel {
     }));
   }
 
+  @NonNull Flowable<Long> getMarkReadRequests() {
+    Flowable<ThreadCountAggregator> nonInitialThreadCount = threadCountStore.getStateFlowable().filter(count -> !(count instanceof ThreadCountAggregator.Init)).take(1);
+
+    return Flowable.combineLatest(markReadRequestPublisher.onBackpressureBuffer(), nonInitialThreadCount, (time, count) -> time);
+  }
+
+  @NonNull Flowable<Integer> getThreadUnreadCount() {
+    return threadCountStore.getStateFlowable().map(ThreadCountAggregator::getCount);
+  }
+
   @NonNull Flowable<ConversationState> getConversationState() {
     return conversationStateStore.getStateFlowable().observeOn(AndroidSchedulers.mainThread());
   }

app/src/main/java/org/thoughtcrime/securesms/conversation/LastSeenHeader.java

@@ -16,12 +16,18 @@ class LastSeenHeader extends StickyHeaderDecoration {
   private final ConversationAdapter adapter;
   private final long                lastSeenTimestamp;
 
+  private long unreadCount;
+
   LastSeenHeader(ConversationAdapter adapter, long lastSeenTimestamp) {
     super(adapter, false, false, ConversationAdapter.HEADER_TYPE_LAST_SEEN);
     this.adapter               = adapter;
     this.lastSeenTimestamp     = lastSeenTimestamp;
   }
 
+  public void setUnreadCount(long unreadCount) {
+    this.unreadCount = unreadCount;
+  }
+
   @Override
   protected boolean hasHeader(RecyclerView parent, StickyHeaderAdapter stickyAdapter, int position) {
     if (lastSeenTimestamp <= 0) {
@@ -42,7 +48,7 @@ class LastSeenHeader extends StickyHeaderDecoration {
   @Override
   protected @NonNull RecyclerView.ViewHolder getHeader(RecyclerView parent, StickyHeaderAdapter stickyAdapter, int position) {
     StickyHeaderViewHolder viewHolder = new StickyHeaderViewHolder(LayoutInflater.from(parent.getContext()).inflate(R.layout.conversation_item_last_seen, parent, false));
-    adapter.onBindLastSeenViewHolder(viewHolder, position);
+    adapter.onBindLastSeenViewHolder(viewHolder, unreadCount);
 
     int widthSpec  = View.MeasureSpec.makeMeasureSpec(parent.getWidth(), View.MeasureSpec.EXACTLY);
     int heightSpec = View.MeasureSpec.makeMeasureSpec(parent.getHeight(), View.MeasureSpec.UNSPECIFIED);

app/src/main/java/org/thoughtcrime/securesms/conversation/ThreadCountAggregator.kt

@@ -0,0 +1,70 @@
+package org.thoughtcrime.securesms.conversation
+
+import org.thoughtcrime.securesms.database.model.ThreadRecord
+
+/**
+ * Describes and aggregates the thread count for a particular thread, for use in the "Last Seen" header.
+ */
+sealed class ThreadCountAggregator {
+
+  abstract val count: Int
+
+  abstract fun updateWith(record: ThreadRecord): ThreadCountAggregator
+
+  /**
+   * The Init object, used as an initial state and returned whenever the given record is an outgoing message.
+   * The conversation fragment already properly cleans up the header when an outgoing message is emitted, so
+   * there's no need to worry about seeing a "zero."
+   */
+  object Init : ThreadCountAggregator() {
+    override val count: Int = 0
+
+    override fun updateWith(record: ThreadRecord): ThreadCountAggregator {
+      return when {
+        record.isOutgoing -> Outgoing
+        else -> Count(record.threadId, record.unreadCount, record.date)
+      }
+    }
+  }
+
+  /**
+   * The Outgoing object, returned whenever the given record is an outgoing message.
+   * The conversation fragment already properly cleans up the header when an outgoing message is emitted, so
+   * there's no need to worry about seeing a "zero."
+   */
+  object Outgoing : ThreadCountAggregator() {
+    override val count: Int = 0
+
+    override fun updateWith(record: ThreadRecord): ThreadCountAggregator {
+      return when {
+        record.isOutgoing -> Outgoing
+        else -> Count(record.threadId, record.unreadCount, record.date)
+      }
+    }
+  }
+
+  /**
+   * Represents an actual count. We keep record of the id and date to use in comparisons with future
+   * ThreadRecord objects.
+   */
+  class Count(val threadId: Long, val unreadCount: Int, val threadDate: Long) : ThreadCountAggregator() {
+    override val count: Int = unreadCount
+
+    /**
+     * "Ratchets" the count to the new state.
+     * * Outgoing records will always result in Empty
+     * * Mismatched threadIds will always create a new Count, initialized with the new thread
+     * * Matching dates will be ignored, as this means that there was no actual change.
+     * * Otherwise, we'll proceed with the new date and aggregate the count.
+     */
+    override fun updateWith(record: ThreadRecord): ThreadCountAggregator {
+      return when {
+        record.isOutgoing -> Outgoing
+        threadId != record.threadId -> Init.updateWith(record)
+        threadDate >= record.date -> this
+        record.unreadCount > 1 -> Init.updateWith(record)
+        else -> Count(threadId, unreadCount + 1, record.date)
+      }
+    }
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/database/ThreadDatabase.java

@@ -521,6 +521,18 @@ public class ThreadDatabase extends Database {
     return getUnreadThreadIdAggregate(SqlUtil.COUNT, cursor -> CursorUtil.getAggregateOrDefault(cursor, 0L, cursor::getLong));
   }
 
+  public long getUnreadMessageCount(long threadId) {
+    SQLiteDatabase db = databaseHelper.getSignalReadableDatabase();
+
+    try (Cursor cursor = db.query(TABLE_NAME, SqlUtil.buildArgs(UNREAD_COUNT), ID_WHERE, SqlUtil.buildArgs(threadId), null, null, null)) {
+      if (cursor.moveToFirst()) {
+        return CursorUtil.requireLong(cursor, UNREAD_COUNT);
+      } else {
+        return 0L;
+      }
+    }
+  }
+
   public @Nullable String getUnreadThreadIdList() {
     return getUnreadThreadIdAggregate(SqlUtil.buildArgs("GROUP_CONCAT(" + ID + ")"),
                                       cursor -> CursorUtil.getAggregateOrDefault(cursor, null, cursor::getString));

app/src/test/java/org/thoughtcrime/securesms/conversation/ThreadCountTest.kt

@@ -0,0 +1,120 @@
+package org.thoughtcrime.securesms.conversation
+
+import org.junit.Assert.assertEquals
+import org.junit.Test
+import org.thoughtcrime.securesms.database.MmsSmsColumns
+import org.thoughtcrime.securesms.database.model.ThreadRecord
+
+class ThreadCountTest {
+
+  @Test
+  fun `Given an Init, when I getCount, then I expect 0`() {
+    // GIVEN
+    val threadCount = ThreadCountAggregator.Init
+
+    // WHEN
+    val result = threadCount.count
+
+    // THEN
+    assertEquals(0, result)
+  }
+
+  @Test
+  fun `Given an Empty, when I updateWith an outgoing record, then I expect Outgoing`() {
+    // GIVEN
+    val threadRecord = createThreadRecord(isOutgoing = true)
+
+    // WHEN
+    val result = ThreadCountAggregator.Init.updateWith(threadRecord)
+
+    // THEN
+    assertEquals(result, ThreadCountAggregator.Outgoing)
+  }
+
+  @Test
+  fun `Given an Empty, when I updateWith an incoming record, then I expect 5`() {
+    // GIVEN
+    val threadRecord = createThreadRecord(unreadCount = 5)
+
+    // WHEN
+    val result = ThreadCountAggregator.Init.updateWith(threadRecord)
+
+    // THEN
+    assertEquals(5, result.count)
+  }
+
+  @Test
+  fun `Given a Count, when I updateWith an incoming record with the same date, then I expect 5`() {
+    // GIVEN
+    val threadRecord = createThreadRecord(unreadCount = 5)
+    val newThreadRecord = createThreadRecord(unreadCount = 1)
+
+    // WHEN
+    val result = ThreadCountAggregator.Init.updateWith(threadRecord).updateWith(newThreadRecord)
+
+    // THEN
+    assertEquals(5, result.count)
+  }
+
+  @Test
+  fun `Given a Count, when I updateWith an incoming record with an earlier date, then I expect 5`() {
+    // GIVEN
+    val threadRecord = createThreadRecord(unreadCount = 5)
+    val newThreadRecord = createThreadRecord(unreadCount = 1, date = 0L)
+
+    // WHEN
+    val result = ThreadCountAggregator.Init.updateWith(threadRecord).updateWith(newThreadRecord)
+
+    // THEN
+    assertEquals(5, result.count)
+  }
+
+  @Test
+  fun `Given a Count, when I updateWith an incoming record with a later date, then I expect 6`() {
+    // GIVEN
+    val threadRecord = createThreadRecord(unreadCount = 5)
+    val newThreadRecord = createThreadRecord(unreadCount = 1, date = 2L)
+
+    // WHEN
+    val result = ThreadCountAggregator.Init.updateWith(threadRecord).updateWith(newThreadRecord)
+
+    // THEN
+    assertEquals(6, result.count)
+  }
+
+  @Test
+  fun `Given a Count, when I updateWith an incoming record with a later date and unread count gt 1, then I expect new unread count`() {
+    // GIVEN
+    val threadRecord = createThreadRecord(unreadCount = 5)
+    val newThreadRecord = createThreadRecord(unreadCount = 3, date = 2L)
+
+    // WHEN
+    val result = ThreadCountAggregator.Init.updateWith(threadRecord).updateWith(newThreadRecord)
+
+    // THEN
+    assertEquals(3, result.count)
+  }
+
+  @Test
+  fun `Given a Count, when I updateWith an incoming record with a different id, then I expect 3`() {
+    // GIVEN
+    val threadRecord = createThreadRecord(threadId = 1L, unreadCount = 5)
+    val newThreadRecord = createThreadRecord(threadId = 2L, unreadCount = 3)
+
+    // WHEN
+    val result = ThreadCountAggregator.Init.updateWith(threadRecord).updateWith(newThreadRecord)
+
+    // THEN
+    assertEquals(3, result.count)
+  }
+
+  private fun createThreadRecord(threadId: Long = 1L, unreadCount: Int = 0, date: Long = 1L, isOutgoing: Boolean = false): ThreadRecord {
+    val outgoingMessageType = MmsSmsColumns.Types.getOutgoingEncryptedMessageType()
+
+    return ThreadRecord.Builder(threadId)
+      .setUnreadCount(unreadCount)
+      .setDate(date)
+      .setType(if (isOutgoing) outgoingMessageType else (outgoingMessageType.inv()))
+      .build()
+  }
+}