Skip to content
This repository was archived by the owner on Apr 26, 2024. It is now read-only.

Commit c677782

Browse files
clokepclokep
committed
Store the thread ID with the receipt.
1 parent 54c9662 commit c677782

File tree

8 files changed

+99
-29
lines changed

8 files changed

+99
-29
lines changed

‎synapse/handlers/receipts.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,7 @@ async def _handle_new_receipts(self, receipts: List[ReadReceipt]) -> bool:
124124
receipt.receipt_type,
125125
receipt.user_id,
126126
receipt.event_ids,
127+
receipt.thread_id,
127128
receipt.data,
128129
)
129130

‎synapse/storage/database.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,8 @@
9595
"local_media_repository_thumbnails": "local_media_repository_thumbnails_method_idx",
9696
"remote_media_cache_thumbnails": "remote_media_repository_thumbnails_method_idx",
9797
"event_push_summary": "event_push_summary_unique_index",
98+
"receipts_linearized": "receipts_linearized_unique_index",
99+
"receipts_graph": "receipts_graph_unique_index",
98100
}
99101

100102

‎synapse/storage/databases/main/receipts.py‎

Lines changed: 53 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -637,6 +637,7 @@ def _insert_linearized_receipt_txn(
637637
receipt_type: str,
638638
user_id: str,
639639
event_id: str,
640+
thread_id: Optional[str],
640641
data: JsonDict,
641642
stream_id: int,
642643
) -> Optional[int]:
@@ -663,12 +664,27 @@ def _insert_linearized_receipt_txn(
663664
# We don't want to clobber receipts for more recent events, so we
664665
# have to compare orderings of existing receipts
665666
if stream_ordering is not None:
666-
sql = (
667-
"SELECT stream_ordering, event_id FROM events"
668-
" INNER JOIN receipts_linearized AS r USING (event_id, room_id)"
669-
" WHERE r.room_id = ? AND r.receipt_type = ? AND r.user_id = ?"
667+
if thread_id is None:
668+
thread_clause = "r.thread_id IS NULL"
669+
thread_args: Tuple[str, ...] = ()
670+
else:
671+
thread_clause = "r.thread_id = ?"
672+
thread_args = (thread_id,)
673+
674+
sql = f"""
675+
SELECT stream_ordering, event_id FROM events
676+
INNER JOIN receipts_linearized AS r USING (event_id, room_id)
677+
WHERE r.room_id = ? AND r.receipt_type = ? AND r.user_id = ? AND {thread_clause}
678+
"""
679+
txn.execute(
680+
sql,
681+
(
682+
room_id,
683+
receipt_type,
684+
user_id,
685+
)
686+
+ thread_args,
670687
)
671-
txn.execute(sql, (room_id, receipt_type, user_id))
672688

673689
for so, eid in txn:
674690
if int(so) >= stream_ordering:
@@ -688,21 +704,28 @@ def _insert_linearized_receipt_txn(
688704
self._receipts_stream_cache.entity_has_changed, room_id, stream_id
689705
)
690706

707+
keyvalues = {
708+
"room_id": room_id,
709+
"receipt_type": receipt_type,
710+
"user_id": user_id,
711+
}
712+
where_clause = ""
713+
if thread_id is None:
714+
where_clause = "thread_id IS NULL"
715+
else:
716+
keyvalues["thread_id"] = thread_id
717+
691718
self.db_pool.simple_upsert_txn(
692719
txn,
693720
table="receipts_linearized",
694-
keyvalues={
695-
"room_id": room_id,
696-
"receipt_type": receipt_type,
697-
"user_id": user_id,
698-
},
721+
keyvalues=keyvalues,
699722
values={
700723
"stream_id": stream_id,
701724
"event_id": event_id,
702725
"event_stream_ordering": stream_ordering,
703726
"data": json_encoder.encode(data),
704-
"thread_id": None,
705727
},
728+
where_clause=where_clause,
706729
# receipts_linearized has a unique constraint on
707730
# (user_id, room_id, receipt_type), so no need to lock
708731
lock=False,
@@ -754,6 +777,7 @@ async def insert_receipt(
754777
receipt_type: str,
755778
user_id: str,
756779
event_ids: List[str],
780+
thread_id: Optional[str],
757781
data: dict,
758782
) -> Optional[Tuple[int, int]]:
759783
"""Insert a receipt, either from local client or remote server.
@@ -786,6 +810,7 @@ async def insert_receipt(
786810
receipt_type,
787811
user_id,
788812
linearized_event_id,
813+
thread_id,
789814
data,
790815
stream_id=stream_id,
791816
# Read committed is actually beneficial here because we check for a receipt with
@@ -800,7 +825,8 @@ async def insert_receipt(
800825

801826
now = self._clock.time_msec()
802827
logger.debug(
803-
"RR for event %s in %s (%i ms old)",
828+
"Receipt %s for event %s in %s (%i ms old)",
829+
receipt_type,
804830
linearized_event_id,
805831
room_id,
806832
now - event_ts,
@@ -813,6 +839,7 @@ async def insert_receipt(
813839
receipt_type,
814840
user_id,
815841
event_ids,
842+
thread_id,
816843
data,
817844
)
818845

@@ -827,6 +854,7 @@ def _insert_graph_receipt_txn(
827854
receipt_type: str,
828855
user_id: str,
829856
event_ids: List[str],
857+
thread_id: Optional[str],
830858
data: JsonDict,
831859
) -> None:
832860
assert self._can_write_to_receipts
@@ -838,19 +866,26 @@ def _insert_graph_receipt_txn(
838866
# FIXME: This shouldn't invalidate the whole cache
839867
txn.call_after(self._get_linearized_receipts_for_room.invalidate, (room_id,))
840868

869+
keyvalues = {
870+
"room_id": room_id,
871+
"receipt_type": receipt_type,
872+
"user_id": user_id,
873+
}
874+
where_clause = ""
875+
if thread_id is None:
876+
where_clause = "thread_id IS NULL"
877+
else:
878+
keyvalues["thread_id"] = thread_id
879+
841880
self.db_pool.simple_upsert_txn(
842881
txn,
843882
table="receipts_graph",
844-
keyvalues={
845-
"room_id": room_id,
846-
"receipt_type": receipt_type,
847-
"user_id": user_id,
848-
},
883+
keyvalues=keyvalues,
849884
values={
850885
"event_ids": json_encoder.encode(event_ids),
851886
"data": json_encoder.encode(data),
852-
"thread_id": None,
853887
},
888+
where_clause=where_clause,
854889
# receipts_graph has a unique constraint on
855890
# (user_id, room_id, receipt_type), so no need to lock
856891
lock=False,

‎tests/handlers/test_appservice.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -447,6 +447,7 @@ def test_sending_read_receipt_batches_to_application_services(self):
447447
receipt_type="m.read",
448448
user_id=self.local_user,
449449
event_ids=[f"$eventid_{i}"],
450+
thread_id=None,
450451
data={},
451452
)
452453
)

‎tests/replication/slave/storage/test_events.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -171,7 +171,7 @@ def test_push_actions_for_user(self, send_receipt: bool):
171171
if send_receipt:
172172
self.get_success(
173173
self.master_store.insert_receipt(
174-
ROOM_ID, ReceiptTypes.READ, USER_ID_2, [event1.event_id], {}
174+
ROOM_ID, ReceiptTypes.READ, USER_ID_2, [event1.event_id], None, {}
175175
)
176176
)
177177

‎tests/replication/tcp/streams/test_receipts.py‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,12 @@ def test_receipt(self):
3333
# tell the master to send a new receipt
3434
self.get_success(
3535
self.hs.get_datastores().main.insert_receipt(
36-
"!room:blue", "m.read", USER_ID, ["$event:blue"], {"a": 1}
36+
"!room:blue",
37+
"m.read",
38+
USER_ID,
39+
["$event:blue"],
40+
thread_id=None,
41+
data={"a": 1},
3742
)
3843
)
3944
self.replicate()
@@ -58,7 +63,12 @@ def test_receipt(self):
5863

5964
self.get_success(
6065
self.hs.get_datastores().main.insert_receipt(
61-
"!room2:blue", "m.read", USER_ID, ["$event2:foo"], {"a": 2}
66+
"!room2:blue",
67+
"m.read",
68+
USER_ID,
69+
["$event2:foo"],
70+
thread_id=None,
71+
data={"a": 2},
6272
)
6373
)
6474
self.replicate()

‎tests/storage/test_event_push_actions.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,7 @@ def _mark_read(event_id: str) -> None:
106106
"m.read",
107107
user_id=user_id,
108108
event_ids=[event_id],
109+
thread_id=None,
109110
data={},
110111
)
111112
)

‎tests/storage/test_receipts.py‎

Lines changed: 28 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -131,13 +131,18 @@ def test_get_receipts_for_user(self) -> None:
131131
# Send public read receipt for the first event
132132
self.get_success(
133133
self.store.insert_receipt(
134-
self.room_id1, ReceiptTypes.READ, OUR_USER_ID, [event1_1_id], {}
134+
self.room_id1, ReceiptTypes.READ, OUR_USER_ID, [event1_1_id], None, {}
135135
)
136136
)
137137
# Send private read receipt for the second event
138138
self.get_success(
139139
self.store.insert_receipt(
140-
self.room_id1, ReceiptTypes.READ_PRIVATE, OUR_USER_ID, [event1_2_id], {}
140+
self.room_id1,
141+
ReceiptTypes.READ_PRIVATE,
142+
OUR_USER_ID,
143+
[event1_2_id],
144+
None,
145+
{},
141146
)
142147
)
143148

@@ -164,7 +169,7 @@ def test_get_receipts_for_user(self) -> None:
164169
# Test receipt updating
165170
self.get_success(
166171
self.store.insert_receipt(
167-
self.room_id1, ReceiptTypes.READ, OUR_USER_ID, [event1_2_id], {}
172+
self.room_id1, ReceiptTypes.READ, OUR_USER_ID, [event1_2_id], None, {}
168173
)
169174
)
170175
res = self.get_success(
@@ -180,7 +185,12 @@ def test_get_receipts_for_user(self) -> None:
180185
# Test new room is reflected in what the method returns
181186
self.get_success(
182187
self.store.insert_receipt(
183-
self.room_id2, ReceiptTypes.READ_PRIVATE, OUR_USER_ID, [event2_1_id], {}
188+
self.room_id2,
189+
ReceiptTypes.READ_PRIVATE,
190+
OUR_USER_ID,
191+
[event2_1_id],
192+
None,
193+
{},
184194
)
185195
)
186196
res = self.get_success(
@@ -202,13 +212,18 @@ def test_get_last_receipt_event_id_for_user(self) -> None:
202212
# Send public read receipt for the first event
203213
self.get_success(
204214
self.store.insert_receipt(
205-
self.room_id1, ReceiptTypes.READ, OUR_USER_ID, [event1_1_id], {}
215+
self.room_id1, ReceiptTypes.READ, OUR_USER_ID, [event1_1_id], None, {}
206216
)
207217
)
208218
# Send private read receipt for the second event
209219
self.get_success(
210220
self.store.insert_receipt(
211-
self.room_id1, ReceiptTypes.READ_PRIVATE, OUR_USER_ID, [event1_2_id], {}
221+
self.room_id1,
222+
ReceiptTypes.READ_PRIVATE,
223+
OUR_USER_ID,
224+
[event1_2_id],
225+
None,
226+
{},
212227
)
213228
)
214229

@@ -241,7 +256,7 @@ def test_get_last_receipt_event_id_for_user(self) -> None:
241256
# Test receipt updating
242257
self.get_success(
243258
self.store.insert_receipt(
244-
self.room_id1, ReceiptTypes.READ, OUR_USER_ID, [event1_2_id], {}
259+
self.room_id1, ReceiptTypes.READ, OUR_USER_ID, [event1_2_id], None, {}
245260
)
246261
)
247262
res = self.get_success(
@@ -259,7 +274,12 @@ def test_get_last_receipt_event_id_for_user(self) -> None:
259274
# Test new room is reflected in what the method returns
260275
self.get_success(
261276
self.store.insert_receipt(
262-
self.room_id2, ReceiptTypes.READ_PRIVATE, OUR_USER_ID, [event2_1_id], {}
277+
self.room_id2,
278+
ReceiptTypes.READ_PRIVATE,
279+
OUR_USER_ID,
280+
[event2_1_id],
281+
None,
282+
{},
263283
)
264284
)
265285
res = self.get_success(

0 commit comments

Comments
 (0)