Take an optional unthreaded receipt position in _get_unread_counts_by_pos_txn

The caller passed either the user's latest unthreaded receipt or, when they
had none, the stream ordering of their join, leaving the callee unable to tell
the two apart. Pass `None` instead and do the membership lookup in the callee,
so the fallback is explicit.

(cherry picked from commit e08b4dabdb)

synapse/storage/databases/main/event_push_actions.py

@@ -533,24 +533,8 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             receipt_types=(ReceiptTypes.READ, ReceiptTypes.READ_PRIVATE),
         )
 
-        if result:
-            _, stream_ordering = result
-
-        else:
-            # If the user has no receipts in the room, retrieve the stream ordering for
-            # the latest membership event from this user in this room (which we assume is
-            # a join).
-            event_id = self.db_pool.simple_select_one_onecol_txn(
-                txn=txn,
-                table="local_current_membership",
-                keyvalues={"room_id": room_id, "user_id": user_id},
-                retcol="event_id",
-            )
-
-            stream_ordering = self.get_stream_id_for_event_txn(txn, event_id)
-
         return self._get_unread_counts_by_pos_txn(
-            txn, room_id, user_id, stream_ordering
+            txn, room_id, user_id, result[1] if result else None
         )
 
     def _get_unread_counts_by_pos_txn(
@@ -558,7 +542,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
         txn: LoggingTransaction,
         room_id: str,
         user_id: str,
-        unthreaded_receipt_stream_ordering: int,
+        unthreaded_receipt_stream_ordering: int | None,
     ) -> RoomNotifCounts:
         """Get the number of unread messages for a user/room that have happened
         since the given stream ordering.
@@ -568,8 +552,8 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             room_id: The room ID to get unread counts for.
             user_id: The user ID to get unread counts for.
             unthreaded_receipt_stream_ordering: The stream ordering of the user's latest
-                unthreaded receipt in the room. If there are no unthreaded receipts,
-                the stream ordering of the user's join event.
+                unthreaded receipt in the room, or None if they have none. In that case
+                we count from the user's latest membership event instead.
 
         Returns:
             A RoomNotifCounts object containing the notification count, the
@@ -577,6 +561,21 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             and threads.
         """
 
+        if unthreaded_receipt_stream_ordering is not None:
+            from_stream_ordering = unthreaded_receipt_stream_ordering
+        else:
+            # If the user has no unthreaded receipt in the room, retrieve the stream
+            # ordering for the latest membership event from this user in this room
+            # (which we assume is a join).
+            event_id = self.db_pool.simple_select_one_onecol_txn(
+                txn=txn,
+                table="local_current_membership",
+                keyvalues={"room_id": room_id, "user_id": user_id},
+                retcol="event_id",
+            )
+
+            from_stream_ordering = self.get_stream_id_for_event_txn(txn, event_id)
+
         main_counts = NotifCounts()
         thread_counts: dict[str, NotifCounts] = {}
 
@@ -631,12 +630,12 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             (
                 user_id,
                 room_id,
-                unthreaded_receipt_stream_ordering,
+                from_stream_ordering,
                 *receipts_args,
                 room_id,
                 user_id,
-                unthreaded_receipt_stream_ordering,
-                unthreaded_receipt_stream_ordering,
+                from_stream_ordering,
+                from_stream_ordering,
             ),
         )
         summarised_threads = set()
@@ -670,11 +669,11 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             (
                 user_id,
                 room_id,
-                unthreaded_receipt_stream_ordering,
+                from_stream_ordering,
                 *receipts_args,
                 user_id,
                 room_id,
-                unthreaded_receipt_stream_ordering,
+                from_stream_ordering,
             ),
         )
         for highlight_count, thread_id in txn:
@@ -748,11 +747,11 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             (
                 user_id,
                 room_id,
-                unthreaded_receipt_stream_ordering,
+                from_stream_ordering,
                 *receipts_args,
                 user_id,
                 room_id,
-                unthreaded_receipt_stream_ordering,
+                from_stream_ordering,
                 *thread_id_args,
             ),
         )