Mark events as read using threaded read receipts from MSC3771

[clokep]

Builds on #13782 & #13776 to handle threaded read receipts (which are now stored in the database) when marking events as read.

As MSC3771 discusses, this handles read receipts of two types:

1. Unthreaded read receipts (aka how read receipts currently work).
2. Threaded read receipts (which operate only on events in the same thread).

Part of #12550.

[erikjohnston]

I think this looks good, I'm a bit nervous about the queries I commented on so would love to know if we could somehow bound them.

[erikjohnston]

I think this query is going to load all receipts in a room? Can we add bound it somehow, like adding a stream_id clause or something?

[clokep]

Yeah, they'll load everything in the room. Will think about how we can limit it!

[clokep]

We can limit this to only fetch threaded receipts after the unthreaded receipt, I think? If the threaded receipt is earlier than the unthreaded receipt then it doesn't matter. (This would essentially replace the GREATEST(threaded_receipt, unthreaded_receipt)?)

[clokep]

@erikjohnston How does 041fe7f look to you?

[erikjohnston]

Yeah, I think that looks good. Do you have an example query plan?

[clokep]

Sorry, I meant to include that, this an EXPLAIN ANALYZE of this query on Matrix HQ with a user. There aren't really currently threaded read receipts in this room though so, maybe this isn't useful?

Matrix HQ

 Hash Left Join  (cost=12.49..16.52 rows=1 width=21) (actual time=1.619..1.799 rows=1 loops=1)
   Hash Cond: (event_push_summary.thread_id = receipts_linearized.thread_id)
   Filter: (((event_push_summary.last_receipt_stream_ordering IS NULL) AND (event_push_summary.stream_ordering > COALESCE((max(events.stream_ordering)), '3156421630'::bigint))) OR (event_push_summary.last_receipt_stream_ordering = COALESCE((max(events.
stream_ordering)), '3156421630'::bigint)))
   ->  Index Scan using event_push_summary_unique_index2 on event_push_summary  (cost=0.56..4.58 rows=1 width=37) (actual time=0.569..0.746 rows=1 loops=1)
         Index Cond: ((user_id = '@e:matrix.org'::text) AND (room_id = '!OGEhHVWSdvArJzumhm:matrix.org'::text))
         Filter: ((notif_count <> 0) OR (COALESCE(unread_count, '0'::bigint) <> 0))
   ->  Hash  (cost=11.92..11.92 rows=1 width=40) (actual time=1.036..1.037 rows=0 loops=1)
         Buckets: 1024  Batches: 1  Memory Usage: 8kB
         ->  GroupAggregate  (cost=11.89..11.91 rows=1 width=40) (actual time=1.036..1.037 rows=0 loops=1)
               Group Key: receipts_linearized.thread_id
               ->  Sort  (cost=11.89..11.90 rows=1 width=40) (actual time=1.034..1.035 rows=0 loops=1)
                     Sort Key: receipts_linearized.thread_id
                     Sort Method: quicksort  Memory: 25kB
                     ->  Nested Loop  (cost=1.27..11.88 rows=1 width=40) (actual time=1.027..1.027 rows=0 loops=1)
                           ->  Index Scan using receipts_linearized_uniqueness_thread on receipts_linearized  (cost=0.56..7.16 rows=1 width=104) (actual time=0.324..0.336 rows=1 loops=1)
                                 Index Cond: ((room_id = '!OGEhHVWSdvArJzumhm:matrix.org'::text) AND (receipt_type = ANY ('{m.read,m.read.private}'::text[])) AND (user_id = '@e:matrix.org'::text))
                           ->  Index Scan using events_event_id_key on events  (cost=0.70..4.73 rows=1 width=78) (actual time=0.687..0.687 rows=0 loops=1)
                                 Index Cond: (event_id = receipts_linearized.event_id)
                                 Filter: ((stream_ordering > '3156421630'::bigint) AND (room_id = '!OGEhHVWSdvArJzumhm:matrix.org'::text))
                                 Rows Removed by Filter: 1
 Planning time: 1.432 ms
 Execution time: 1.892 ms
(22 rows)

I also ran one with some threaded read receipts:

Test room w/ threaded receipts

 Hash Left Join  (cost=26.83..36.92 rows=1 width=56) (actual time=0.030..0.032 rows=0 loops=1)
   Hash Cond: (event_push_summary.thread_id = receipts_linearized.thread_id)
   Filter: (((event_push_summary.last_receipt_stream_ordering IS NULL) AND (event_push_summary.stream_ordering > COALESCE
((max(events.stream_ordering)), '1856'::bigint))) OR (event_push_summary.last_receipt_stream_ordering = COALESCE((max(events.stream_ordering)), '1856'::bigint)))
   ->  Bitmap Heap Scan on event_push_summary  (cost=4.31..14.39 rows=4 width=72) (actual time=0.030..0.030 rows=0 loops=1)
         Recheck Cond: ((user_id = '@clokep:threads-dev.lab.element.dev'::text) AND (room_id = '!BCtxSvqOvcQQYhkLsB:threads-dev.lab.element.dev'::text))
         Filter: ((notif_count <> 0) OR (COALESCE(unread_count, '0'::bigint) <> 0))
         Rows Removed by Filter: 4
         Heap Blocks: exact=2
         ->  Bitmap Index Scan on event_push_summary_unique_index2  (cost=0.00..4.31 rows=4 width=0) (actual time=0.018..0.018 rows=4 loops=1)
               Index Cond: ((user_id = '@clokep:threads-dev.lab.element.dev'::text) AND (room_id = '!BCtxSvqOvcQQYhkLsB:threads-dev.lab.element.dev'::text))
   ->  Hash  (cost=22.50..22.50 rows=1 width=24) (never executed)
         ->  GroupAggregate  (cost=22.47..22.49 rows=1 width=24) (never executed)
               Group Key: receipts_linearized.thread_id
               ->  Sort  (cost=22.47..22.48 rows=1 width=24) (never executed)
                     Sort Key: receipts_linearized.thread_id
                     ->  Nested Loop  (cost=4.88..22.46 rows=1 width=24) (never executed)
                           Join Filter: (receipts_linearized.event_id = events.event_id)
                           ->  Index Scan using events_stream_ordering on events  (cost=0.28..7.83 rows=1 width=87) (never executed)
                                 Index Cond: (stream_ordering > '3156421630'::bigint)
                                 Filter: (room_id = '!BCtxSvqOvcQQYhkLsB:threads-dev.lab.element.dev'::text)
                           ->  Bitmap Heap Scan on receipts_linearized  (cost=4.59..14.57 rows=5 width=100) (never executed)
                                 Recheck Cond: ((room_id = '!BCtxSvqOvcQQYhkLsB:threads-dev.lab.element.dev'::text) AND (user_id = '@clokep:threads-dev.lab.element.dev'::text))
                                 Filter: (receipt_type = ANY ('{m.read,m.read.private}'::text[]))
                                 ->  Bitmap Index Scan on receipts_linearized_uniqueness_thread  (cost=0.00..4.59 rows=5 width=0) (never executed)
                                       Index Cond: ((room_id = '!BCtxSvqOvcQQYhkLsB:threads-dev.lab.element.dev'::text) AND (user_id = '@clokep:threads-dev.lab.element.dev'::text))
 Planning Time: 1.027 ms
 Execution Time: 0.121 ms
(27 rows)