Check for all receipt types

Signed-off-by: Šimon Brandner <[email protected]>

synapse/storage/databases/main/event_push_actions.py

@@ -19,7 +19,7 @@
 
 from synapse.api.constants import ReceiptTypes
 from synapse.metrics.background_process_metrics import wrap_as_background_process
-from synapse.storage._base import SQLBaseStore, db_to_json
+from synapse.storage._base import SQLBaseStore, db_to_json, make_in_list_sql_clause
 from synapse.storage.database import (
     DatabasePool,
     LoggingDatabaseConnection,
@@ -380,35 +380,49 @@ async def get_unread_push_actions_for_user_in_range_for_http(
             The list will be ordered by ascending stream_ordering.
             The list will have between 0~limit entries.
         """
+
+        receipt_types_clause, args = make_in_list_sql_clause(
+            self.database_engine,
+            "receipt_type",
+            (
+                ReceiptTypes.READ,
+                ReceiptTypes.READ_PRIVATE,
+                ReceiptTypes.UNSTABLE_READ_PRIVATE,
+            ),
+        )
+
         # find rooms that have a read receipt in them and return the next
         # push actions
         def get_after_receipt(
             txn: LoggingTransaction,
         ) -> List[Tuple[str, str, int, str, bool]]:
             # find rooms that have a read receipt in them and return the next
             # push actions
-            sql = (
-                "SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions,"
-                "   ep.highlight "
-                " FROM ("
-                "   SELECT room_id,"
-                "       MAX(stream_ordering) as stream_ordering"
-                "   FROM events"
-                "   INNER JOIN receipts_linearized USING (room_id, event_id)"
-                "   WHERE receipt_type = 'm.read' AND user_id = ?"
-                "   GROUP BY room_id"
-                ") AS rl,"
-                " event_push_actions AS ep"
-                " WHERE"
-                "   ep.room_id = rl.room_id"
-                "   AND ep.stream_ordering > rl.stream_ordering"
-                "   AND ep.user_id = ?"
-                "   AND ep.stream_ordering > ?"
-                "   AND ep.stream_ordering <= ?"
-                "   AND ep.notif = 1"
-                " ORDER BY ep.stream_ordering ASC LIMIT ?"
+
+            sql = f"""(
+                SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions,
+                   ep.highlight
+                 FROM (
+                   SELECT room_id,
+                       MAX(stream_ordering) as stream_ordering
+                   FROM events
+                   INNER JOIN receipts_linearized USING (room_id, event_id)
+                   WHERE {receipt_types_clause} AND user_id = ?
+                   GROUP BY room_id
+                ) AS rl,
+                 event_push_actions AS ep
+                 WHERE
+                   ep.room_id = rl.room_id
+                   AND ep.stream_ordering > rl.stream_ordering
+                   AND ep.user_id = ?
+                   AND ep.stream_ordering > ?
+                   AND ep.stream_ordering <= ?
+                   AND ep.notif = 1
+                 ORDER BY ep.stream_ordering ASC LIMIT ?
+            """
+            args.extend(
+                user_id, user_id, min_stream_ordering, max_stream_ordering, limit
             )
-            args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
             txn.execute(sql, args)
             return cast(List[Tuple[str, str, int, str, bool]], txn.fetchall())
 
@@ -422,24 +436,26 @@ def get_after_receipt(
         def get_no_receipt(
             txn: LoggingTransaction,
         ) -> List[Tuple[str, str, int, str, bool]]:
-            sql = (
-                "SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions,"
-                "   ep.highlight "
-                " FROM event_push_actions AS ep"
-                " INNER JOIN events AS e USING (room_id, event_id)"
-                " WHERE"
-                "   ep.room_id NOT IN ("
-                "     SELECT room_id FROM receipts_linearized"
-                "       WHERE receipt_type = 'm.read' AND user_id = ?"
-                "       GROUP BY room_id"
-                "   )"
-                "   AND ep.user_id = ?"
-                "   AND ep.stream_ordering > ?"
-                "   AND ep.stream_ordering <= ?"
-                "   AND ep.notif = 1"
-                " ORDER BY ep.stream_ordering ASC LIMIT ?"
+            sql = f"""
+                SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions,
+                   ep.highlight
+                 FROM event_push_actions AS ep
+                 INNER JOIN events AS e USING (room_id, event_id)
+                 WHERE
+                   ep.room_id NOT IN (
+                     SELECT room_id FROM receipts_linearized
+                       WHERE {receipt_types_clause} AND user_id = ?
+                       GROUP BY room_id
+                   )
+                   AND ep.user_id = ?
+                   AND ep.stream_ordering > ?
+                   AND ep.stream_ordering <= ?
+                   AND ep.notif = 1
+                 ORDER BY ep.stream_ordering ASC LIMIT ?
+            """
+            args.extends(
+                (user_id, user_id, min_stream_ordering, max_stream_ordering, limit)
             )
-            args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
             txn.execute(sql, args)
             return cast(List[Tuple[str, str, int, str, bool]], txn.fetchall())
 
@@ -489,20 +505,31 @@ async def get_unread_push_actions_for_user_in_range_for_email(
             The list will be ordered by descending received_ts.
             The list will have between 0~limit entries.
         """
+
+        receipt_types_clause, args = make_in_list_sql_clause(
+            self.database_engine,
+            "receipt_type",
+            (
+                ReceiptTypes.READ,
+                ReceiptTypes.READ_PRIVATE,
+                ReceiptTypes.UNSTABLE_READ_PRIVATE,
+            ),
+        )
+
         # find rooms that have a read receipt in them and return the most recent
         # push actions
         def get_after_receipt(
             txn: LoggingTransaction,
         ) -> List[Tuple[str, str, int, str, bool, int]]:
-            sql = (
+            sql = f"""
                 "SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions,"
                 "  ep.highlight, e.received_ts"
                 " FROM ("
                 "   SELECT room_id,"
                 "       MAX(stream_ordering) as stream_ordering"
                 "   FROM events"
                 "   INNER JOIN receipts_linearized USING (room_id, event_id)"
-                "   WHERE receipt_type = 'm.read' AND user_id = ?"
+                "   WHERE {receipt_types_clause} AND user_id = ?"
                 "   GROUP BY room_id"
                 ") AS rl,"
                 " event_push_actions AS ep"
@@ -515,8 +542,10 @@ def get_after_receipt(
                 "   AND ep.stream_ordering <= ?"
                 "   AND ep.notif = 1"
                 " ORDER BY ep.stream_ordering DESC LIMIT ?"
+            """
+            args.extends(
+                (user_id, user_id, min_stream_ordering, max_stream_ordering, limit)
             )
-            args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
             txn.execute(sql, args)
             return cast(List[Tuple[str, str, int, str, bool, int]], txn.fetchall())
 
@@ -530,24 +559,26 @@ def get_after_receipt(
         def get_no_receipt(
             txn: LoggingTransaction,
         ) -> List[Tuple[str, str, int, str, bool, int]]:
-            sql = (
-                "SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions,"
-                "   ep.highlight, e.received_ts"
-                " FROM event_push_actions AS ep"
-                " INNER JOIN events AS e USING (room_id, event_id)"
-                " WHERE"
-                "   ep.room_id NOT IN ("
-                "     SELECT room_id FROM receipts_linearized"
-                "       WHERE receipt_type = 'm.read' AND user_id = ?"
-                "       GROUP BY room_id"
-                "   )"
-                "   AND ep.user_id = ?"
-                "   AND ep.stream_ordering > ?"
-                "   AND ep.stream_ordering <= ?"
-                "   AND ep.notif = 1"
-                " ORDER BY ep.stream_ordering DESC LIMIT ?"
+            sql = f"""
+                SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions,
+                   ep.highlight, e.received_ts
+                 FROM event_push_actions AS ep
+                 INNER JOIN events AS e USING (room_id, event_id)
+                 WHERE
+                   ep.room_id NOT IN (
+                     SELECT room_id FROM receipts_linearized
+                       WHERE {receipt_types_clause} AND user_id = ?
+                       GROUP BY room_id
+                   )
+                   AND ep.user_id = ?
+                   AND ep.stream_ordering > ?
+                   AND ep.stream_ordering <= ?
+                   AND ep.notif = 1
+                 ORDER BY ep.stream_ordering DESC LIMIT ?
+            """
+            args.extend(
+                user_id, user_id, min_stream_ordering, max_stream_ordering, limit
             )
-            args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
             txn.execute(sql, args)
             return cast(List[Tuple[str, str, int, str, bool, int]], txn.fetchall())