tuwunel_admin/user/
last_active.rs

1use std::cmp::{self, Reverse};
2
3use futures::{FutureExt, StreamExt, TryStreamExt};
4use ruma::{MilliSecondsSinceUnixEpoch, uint};
5use tuwunel_core::{
6	Result,
7	itertools::Itertools,
8	utils::{ReadyExt, stream::IterStream},
9};
10
11use crate::admin_command;
12
13#[admin_command]
14pub(super) async fn last_active(&self, limit: Option<usize>) -> Result {
15	self.services
16		.users
17		.list_local_users()
18		.map(ToOwned::to_owned)
19		.then(async |user_id| {
20			self.services
21				.users
22				.all_devices_metadata(&user_id)
23				.ready_filter_map(|device| {
24					device
25						.last_seen_ts
26						.map(|ts| (ts, device.last_seen_ip))
27				})
28				.ready_fold((MilliSecondsSinceUnixEpoch(uint!(0)), None), cmp::max)
29				.map(|(last_seen_ts, last_seen_ip)| (last_seen_ts, last_seen_ip, user_id.clone()))
30				.await
31		})
32		.ready_filter(|(ts, ..)| ts.get() > uint!(0))
33		.collect::<Vec<_>>()
34		.map(IntoIterator::into_iter)
35		.map(|iter| iter.sorted_by_key(|k| Reverse(k.0)))
36		.map(IterStream::try_stream)
37		.flatten_stream()
38		.take(limit.unwrap_or(48))
39		.try_for_each(async |(last_seen_ts, last_seen_ip, user_id)| {
40			let ago = last_seen_ts;
41			let user_id = user_id.localpart();
42			let ip = last_seen_ip.as_deref().unwrap_or_default();
43
44			write!(self, "{ago:?} {ip:<40} {user_id}\n").await
45		})
46		.boxed()
47		.await
48}
tuwunel_admin/user/last_active.rs

tuwunel_admin/user/
last_active.rs
