Skip to main content

tuwunel_matrix/room/state/resolution/resolve/
auth_difference.rs

1use std::{borrow::Borrow, collections::HashMap, hash::Hash};
2
3use futures::{FutureExt, Stream};
4use ruma::EventId;
5use tuwunel_core::utils::stream::{IterStream, ReadyExt};
6
7use super::AuthSet;
8use crate::event_id::RandomState;
9
10struct Counts<Id> {
11	// Leading run of input chains holding each ID; a repeat within a chain is a no-op.
12	streaks: HashMap<Id, usize, RandomState>,
13	total: usize,
14}
15
16/// Returns auth events absent from at least one input chain.
17///
18/// The difference is the union of the chains minus their intersection.
19/// Repeated event IDs count once per input chain, and output order is arbitrary.
20#[tracing::instrument(level = "trace", skip_all)]
21pub(super) fn auth_difference<'a, AuthSets, Id>(auth_sets: AuthSets) -> impl Stream<Item = Id>
22where
23	AuthSets: Stream<Item = AuthSet<Id>>,
24	Id: Borrow<EventId> + Clone + Eq + Hash + Send + 'a,
25{
26	auth_sets
27		.ready_fold_default(Counts::merge)
28		.map(|Counts { streaks, total }: Counts<Id>| {
29			streaks
30				.into_iter()
31				.filter_map(move |(id, streak)| streak.lt(&total).then_some(id))
32				.stream()
33		})
34		.flatten_stream()
35}
36
37impl<Id> Default for Counts<Id> {
38	fn default() -> Self { Self { streaks: HashMap::default(), total: 0 } }
39}
40
41impl<Id: Eq + Hash> Counts<Id> {
42	fn merge(mut self, set: AuthSet<Id>) -> Self {
43		self.total = self.total.saturating_add(1);
44		for id in set {
45			let streak = self.streaks.entry(id).or_default();
46
47			if streak.saturating_add(1) == self.total {
48				*streak = self.total;
49			}
50		}
51
52		self
53	}
54}