SocialPub/PrivaPub/Domain/Statistics/StatisticsQueries.cs

using MongoDB.Entities;

using PrivaPub.Infrastructure.Statistics;
using PrivaPub.Models.Jobs;
using PrivaPub.Models.Statistics;

namespace PrivaPub.Domain.Statistics
{
	public sealed record HostDay(DateTime Day, string Host, IReadOnlyDictionary<string, long> Counters, IReadOnlyDictionary<string, long> Latency,
		IReadOnlyDictionary<string, long> Bytes, IReadOnlyDictionary<string, long> Reads, int Accounts, bool Live);

	public class StatisticsQueries
	{
		public const string Attribution = "IP geolocation by DB-IP (https://db-ip.com), CC BY 4.0";
		static readonly string[] Channels = { Interactions.Receive, Interactions.In, Interactions.Out, Interactions.Http, Interactions.Preview, Interactions.Crawl };
		static readonly int[] LatencyLimits = { 50, 100, 250, 500, 1000, 2500, 5000, 10000, 30000 };

		public async Task<List<HostDay>> Days(int days, string host, CancellationToken token)
		{
			var today = DateTime.UtcNow.Date;
			var since = today.AddDays(1 - Math.Clamp(days, 1, 3660));
			var rows = await DB.Default.Find<InstanceDay>()
				.Match(d => d.Day >= since)
				.Match(d => host == null || d.Host == host)
				.ExecuteAsync(token);
			var result = rows.Where(r => r.RolledUpAt != null)
				.Select(r => new HostDay(r.Day, r.Host, r.Counters, r.Latency, r.Bytes, r.Reads, r.Accounts, false))
				.ToList();
			var rolledDays = rows.Where(r => r.RolledUpAt != null).Select(r => r.Day).ToHashSet();
			for (var day = since; day <= today; day = day.AddDays(1))
			{
				if (rolledDays.Contains(day))
					continue;
				var end = day.AddDays(1);
				var start = day;
				var events = await DB.Default.Find<InteractionEvent>()
					.Match(e => e.At >= start && e.At < end)
					.Match(e => host == null || e.Host == host)
					.ExecuteAsync(token);
				var reads = rows.Where(r => r.Day == day).ToDictionary(r => r.Host, r => r.Reads);
				foreach (var (name, fold) in Rollup.Fold(events).Hosts)
					result.Add(new HostDay(day, name, fold.Counters, fold.Latency, fold.Bytes, reads.GetValueOrDefault(name) ?? new(), fold.Accounts.Count, true));
				foreach (var (name, read) in reads.Where(r => result.All(h => h.Day != day || h.Host != r.Key)))
					result.Add(new HostDay(day, name, new Dictionary<string, long>(), new Dictionary<string, long>(), new Dictionary<string, long>(), read, 0, true));
			}
			return result.OrderBy(d => d.Day).ThenBy(d => d.Host, StringComparer.Ordinal).ToList();
		}

		public async Task<object> Overview(int days, CancellationToken token)
		{
			var hostDays = await Days(days, default, token);
			var counters = Sum(hostDays.Select(d => d.Counters));
			var latency = Sum(hostDays.Select(d => d.Latency));
			var server = await DB.Default.Find<ServerDay>().Match(d => d.Day >= DateTime.UtcNow.Date.AddDays(1 - Math.Clamp(days, 1, 3660))).ExecuteAsync(token);
			var delivered = Outcomes(counters, Interactions.Out);
			var attempts = delivered.Where(o => o.Key != Interactions.Deferred).Sum(o => o.Value);
			return new
			{
				Days = days,
				Channels = Channels.ToDictionary(c => c, c => counters.Where(k => k.Key.StartsWith(c + ":", StringComparison.Ordinal)).Sum(k => k.Value)),
				Inbound = Outcomes(counters, Interactions.In),
				Outbound = delivered,
				DeliverySuccess = attempts == 0 ? (double?)null : Math.Round(delivered.GetValueOrDefault(Interactions.Ok) / (double)attempts, 4),
				DeliveryLatencyMs = new { P50 = Percentile(latency, Interactions.Out, 0.5), P95 = Percentile(latency, Interactions.Out, 0.95) },
				Hosts = new
				{
					Active = hostDays.Where(d => d.Host != Interactions.Unknown && d.Counters.Count > 0).Select(d => d.Host).Distinct().Count(),
					Known = await DB.Default.CountAsync<RemoteInstance>(cancellation: token)
				},
				Accounts = server.Sum(d => d.Accounts) + hostDays.Where(d => d.Live).Sum(d => d.Accounts),
				Ledger = new
				{
					Written = server.Sum(d => d.LedgerWritten),
					Dropped = server.Sum(d => d.LedgerDropped),
					Failed = server.Sum(d => d.LedgerFailed)
				}
			};
		}

		public async Task<object> Hosts(int days, string sort, string software, int page, int limit, CancellationToken token)
		{
			var hostDays = await Days(days, default, token);
			var instances = (await DB.Default.Find<RemoteInstance>().ExecuteAsync(token)).ToDictionary(i => i.Host, StringComparer.Ordinal);
			var rows = hostDays.GroupBy(d => d.Host)
				.Select(g =>
				{
					var counters = Sum(g.Select(d => d.Counters));
					var outbound = Outcomes(counters, Interactions.Out);
					var instance = instances.GetValueOrDefault(g.Key);
					return new
					{
						Host = g.Key,
						instance?.Software,
						Version = instance?.SoftwareVersion,
						In = counters.Where(k => k.Key.StartsWith("in:", StringComparison.Ordinal)).Sum(k => k.Value),
						Out = counters.Where(k => k.Key.StartsWith("out:", StringComparison.Ordinal)).Sum(k => k.Value),
						Refused = counters.Where(k => k.Key.StartsWith("recv:", StringComparison.Ordinal) && !k.Key.Contains(":202:", StringComparison.Ordinal)).Sum(k => k.Value),
						Failures = outbound.Where(o => o.Key is Interactions.Retry or Interactions.Dead or Interactions.Failed).Sum(o => o.Value),
						LatencyP50 = Percentile(Sum(g.Select(d => d.Latency)), Interactions.Out, 0.5),
						Accounts = g.Sum(d => d.Accounts),
						instance?.DescribedAt,
						Delivery = instance == default ? default : new { instance.ConsecutiveFailures, instance.UnavailableUntil, instance.LastSuccessAt, instance.LastFailureAt }
					};
				})
				.Where(r => software == default || string.Equals(r.Software, software, StringComparison.OrdinalIgnoreCase));
			rows = sort switch
			{
				"failures" => rows.OrderByDescending(r => r.Failures),
				"latency" => rows.OrderByDescending(r => r.LatencyP50 ?? 0),
				"host" => rows.OrderBy(r => r.Host, StringComparer.Ordinal),
				_ => rows.OrderByDescending(r => r.In + r.Out)
			};
			var list = rows.ToList();
			limit = Math.Clamp(limit, 1, 200);
			return new { Total = list.Count, Page = Math.Max(page, 1), Hosts = list.Skip((Math.Max(page, 1) - 1) * limit).Take(limit), Attribution };
		}

		public async Task<object> Host(string host, int days, CancellationToken token)
		{
			host = Interactions.Host(host);
			var instance = await DB.Default.Find<RemoteInstance>().Match(i => i.Host == host).ExecuteFirstAsync(token);
			var hostDays = await Days(days, host, token);
			var events = await DB.Default.Find<InteractionEvent>().Match(e => e.Host == host).Sort(e => e.At, Order.Descending).Limit(100).ExecuteAsync(token);
			if (instance == default && hostDays.Count == 0 && events.Count == 0)
				return default;
			return new
			{
				Host = host,
				Instance = instance,
				Days = hostDays.Select(d => new { d.Day, d.Counters, d.Latency, d.Bytes, d.Reads, d.Accounts, d.Live }),
				Events = events,
				Attribution
			};
		}

		public async Task<List<InteractionEvent>> Events(string host, string channel, string outcome, string reason, DateTime? before, int limit, CancellationToken token)
		{
			var hostName = host == default ? default : Interactions.Host(host);
			return await DB.Default.Find<InteractionEvent>()
				.Match(e => hostName == null || e.Host == hostName)
				.Match(e => channel == null || e.Channel == channel)
				.Match(e => outcome == null || e.Outcome == outcome)
				.Match(e => reason == null || e.Reason == reason)
				.Match(e => before == null || e.At < before)
				.Sort(e => e.At, Order.Descending)
				.Limit(Math.Clamp(limit, 1, 200))
				.ExecuteAsync(token);
		}

		public async Task<object> Crawler(bool enabled, int revisitDays, CancellationToken token)
		{
			var due = DateTime.UtcNow.AddDays(-Math.Max(revisitDays, 1));
			var recent = await DB.Default.Find<RemoteInstance>()
				.Match(i => i.LastCrawledAt != null)
				.Sort(i => i.LastCrawledAt, Order.Descending)
				.Limit(20)
				.ExecuteAsync(token);
			return new
			{
				Enabled = enabled,
				Touched = await DB.Default.CountAsync<RemoteInstance>(i => i.Seen == "touched", token),
				Crawled = await DB.Default.CountAsync<RemoteInstance>(i => i.Seen == "crawled", token),
				Due = await DB.Default.CountAsync<RemoteInstance>(i => i.LastCrawledAt == null || i.LastCrawledAt < due, token),
				RobotsDisallowed = await DB.Default.CountAsync<RemoteInstance>(i => i.RobotsDisallowed == true, token),
				Recent = recent.Select(i => new { i.Host, i.Seen, i.LastCrawledAt, i.PeersCount, i.RobotsDisallowed, i.CrawlError, i.Software })
			};
		}

		public async Task<List<ServerDay>> Server(int days, CancellationToken token)
		{
			var since = DateTime.UtcNow.Date.AddDays(1 - Math.Clamp(days, 1, 3660));
			return await DB.Default.Find<ServerDay>().Match(d => d.Day >= since).Sort(d => d.Day, Order.Ascending).ExecuteAsync(token);
		}

		public static Dictionary<string, long> Outcomes(IReadOnlyDictionary<string, long> counters, string channel)
		{
			var outcomes = new Dictionary<string, long>();
			foreach (var (key, value) in counters)
			{
				var parts = key.Split(':');
				if (parts.Length >= 4 && parts[0] == channel)
					outcomes[parts[3]] = outcomes.GetValueOrDefault(parts[3]) + value;
			}
			return outcomes;
		}

		public static int? Percentile(IReadOnlyDictionary<string, long> latency, string prefix, double share)
		{
			var buckets = LatencyLimits.Select(limit => (Limit: (int?)limit, Count: latency.GetValueOrDefault($"{prefix}:le{limit}ms")))
				.Append((Limit: (int?)null, Count: latency.GetValueOrDefault($"{prefix}:inf")))
				.ToList();
			var total = buckets.Sum(b => b.Count);
			if (total == 0)
				return default;
			long seen = 0;
			foreach (var (limit, count) in buckets)
			{
				seen += count;
				if (seen >= share * total)
					return limit ?? LatencyLimits[^1] + 1;
			}
			return default;
		}

		static Dictionary<string, long> Sum(IEnumerable<IReadOnlyDictionary<string, long>> dictionaries)
		{
			var sum = new Dictionary<string, long>();
			foreach (var dictionary in dictionaries)
				foreach (var (key, value) in dictionary ?? new Dictionary<string, long>())
					sum[key] = sum.GetValueOrDefault(key) + value;
			return sum;
		}
	}
}