Aggregate unread notif count query for badge count calculation

[Fizzadar]

Attempt to optimise badge count calculation per discussion in #13846 (related: #13448).

Signed off by Nick @ Beeper (@Fizzadar).

Pull Request Checklist

• Pull request is based on the develop branch
• Pull request includes a changelog file.
• Pull request includes a sign off
• Code style is correct
  (run the linters)

[Fizzadar]

Possible missed optimisation here: users with multiple pushers. Perhaps the function should take a stream order, run through a linearizer and use a short TTL cache to avoid double lookups of those.

[clokep]

Sorry for the delay on this, had a couple things come up.

Overall I think this looks reasonable, but I do have a few questions / comments!

[clokep]

Won't this skip any threads which take place wholly after the latest summary? (I'm unsure why we care about whether we've seen the thread or not -- I thought the SQL query was only returning things with valid summaries in the first place?)

[Fizzadar]

The query after this one captures threads that only exist post-summary, this query is specifically to capture items in event_push_actions that do have valid summaries as well. This (all three queries in this method) was based off the thee single room calculations here:

synapse/synapse/storage/databases/main/event_push_actions.py

Lines 586 to 740 in 09de2ae

	# First we pull the counts from the summary table.
	#
	# We check that `last_receipt_stream_ordering` matches the stream ordering of the
	# latest receipt for the thread (which may be either the unthreaded read receipt
	# or the threaded read receipt).
	#
	# If it doesn't match then a new read receipt has arrived and we haven't yet
	# updated the counts in `event_push_summary` to reflect that; in that case we
	# simply ignore `event_push_summary` counts.
	#
	# We then do a manual count of all the rows in the `event_push_actions` table
	# for any user/room/thread which did not have a valid summary found.
	#
	# If `last_receipt_stream_ordering` is null then that means it's up-to-date
	# (as the row was written by an older version of Synapse that
	# updated `event_push_summary` synchronously when persisting a new read
	# receipt).
	txn.execute(
	f"""
	SELECT notif_count, COALESCE(unread_count, 0), thread_id
	FROM event_push_summary
	LEFT JOIN (
	SELECT thread_id, MAX(stream_ordering) AS threaded_receipt_stream_ordering
	FROM receipts_linearized
	LEFT JOIN events USING (room_id, event_id)
	WHERE
	user_id = ?
	AND room_id = ?
	AND stream_ordering > ?
	AND {receipt_types_clause}
	GROUP BY thread_id
	) AS receipts USING (thread_id)
	WHERE room_id = ? AND user_id = ?
	AND (
	(last_receipt_stream_ordering IS NULL AND stream_ordering > COALESCE(threaded_receipt_stream_ordering, ?))
	OR last_receipt_stream_ordering = COALESCE(threaded_receipt_stream_ordering, ?)
	) AND (notif_count != 0 OR COALESCE(unread_count, 0) != 0)
	""",
	(
	user_id,
	room_id,
	unthreaded_receipt_stream_ordering,
	*receipts_args,
	room_id,
	user_id,
	unthreaded_receipt_stream_ordering,
	unthreaded_receipt_stream_ordering,
	),
	)
	summarised_threads = set()
	for notif_count, unread_count, thread_id in txn:
	summarised_threads.add(thread_id)
	counts = _get_thread(thread_id)
	counts.notify_count += notif_count
	counts.unread_count += unread_count
	# Next we need to count highlights, which aren't summarised
	sql = f"""
	SELECT COUNT(*), thread_id FROM event_push_actions
	LEFT JOIN (
	SELECT thread_id, MAX(stream_ordering) AS threaded_receipt_stream_ordering
	FROM receipts_linearized
	LEFT JOIN events USING (room_id, event_id)
	WHERE
	user_id = ?
	AND room_id = ?
	AND stream_ordering > ?
	AND {receipt_types_clause}
	GROUP BY thread_id
	) AS receipts USING (thread_id)
	WHERE user_id = ?
	AND room_id = ?
	AND stream_ordering > COALESCE(threaded_receipt_stream_ordering, ?)
	AND highlight = 1
	GROUP BY thread_id
	"""
	txn.execute(
	sql,
	(
	user_id,
	room_id,
	unthreaded_receipt_stream_ordering,
	*receipts_args,
	user_id,
	room_id,
	unthreaded_receipt_stream_ordering,
	),
	)
	for highlight_count, thread_id in txn:
	_get_thread(thread_id).highlight_count += highlight_count
	# For threads which were summarised we need to count actions since the last
	# rotation.
	thread_id_clause, thread_id_args = make_in_list_sql_clause(
	self.database_engine, "thread_id", summarised_threads
	)
	# The (inclusive) event stream ordering that was previously summarised.
	rotated_upto_stream_ordering = self.db_pool.simple_select_one_onecol_txn(
	txn,
	table="event_push_summary_stream_ordering",
	keyvalues={},
	retcol="stream_ordering",
	)
	unread_counts = self._get_notif_unread_count_for_user_room(
	txn, room_id, user_id, rotated_upto_stream_ordering
	)
	for notif_count, unread_count, thread_id in unread_counts:
	if thread_id not in summarised_threads:
	continue
	if thread_id == MAIN_TIMELINE:
	counts.notify_count += notif_count
	counts.unread_count += unread_count
	elif thread_id in thread_counts:
	thread_counts[thread_id].notify_count += notif_count
	thread_counts[thread_id].unread_count += unread_count
	else:
	# Previous thread summaries of 0 are discarded above.
	#
	# TODO If empty summaries are deleted this can be removed.
	thread_counts[thread_id] = NotifCounts(
	notify_count=notif_count,
	unread_count=unread_count,
	highlight_count=0,
	)
	# Finally we need to count push actions that aren't included in the
	# summary returned above. This might be due to recent events that haven't
	# been summarised yet or the summary is out of date due to a recent read
	# receipt.
	sql = f"""
	SELECT
	COUNT(CASE WHEN notif = 1 THEN 1 END),
	COUNT(CASE WHEN unread = 1 THEN 1 END),
	thread_id
	FROM event_push_actions
	LEFT JOIN (
	SELECT thread_id, MAX(stream_ordering) AS threaded_receipt_stream_ordering
	FROM receipts_linearized
	LEFT JOIN events USING (room_id, event_id)
	WHERE
	user_id = ?
	AND room_id = ?
	AND stream_ordering > ?
	AND {receipt_types_clause}
	GROUP BY thread_id
	) AS receipts USING (thread_id)
	WHERE user_id = ?
	AND room_id = ?
	AND stream_ordering > COALESCE(threaded_receipt_stream_ordering, ?)
	AND NOT {thread_id_clause}
	GROUP BY thread_id
	"""

[Fizzadar]

Apologies for the delay @clokep, this slipped through the net! Have now addressed all the comments I believe!

[clokep]

I think this looks good. 👍

[clokep]

@Fizzadar Thanks so much! I don't think this closes #13846, but it might?