SocialPub/PrivaPub/Federation/Crawler/Crawler.cs

using Microsoft.Extensions.Options;

using MongoDB.Entities;

using PrivaPub.Federation.Moderation;
using PrivaPub.Federation.Objects;
using PrivaPub.Infrastructure.Http;
using PrivaPub.Infrastructure.Jobs;
using PrivaPub.Infrastructure.Statistics;
using PrivaPub.Models;
using PrivaPub.Models.Jobs;
using PrivaPub.Models.Statistics;
using PrivaPub.StaticServices;

namespace PrivaPub.Federation.Crawler
{
	public static class Stargazer
	{
		public const string Token = "PrivaPub-Stargazer";
		public const string Page = "/stargazing";

		public static string UserAgent(string baseAddress) => $"{Token}/{BuildInfo.Ref} (+{baseAddress?.TrimEnd('/')}{Page})";
	}

	//each hour, the servers due for a visit, spread across the hour so the crawler reaches one at a time
	public class CrawlPlanner : IJobHandler
	{
		readonly IJobQueue _queue;
		readonly IDomainBlocks _blocks;
		readonly IOptionsMonitor<StatisticsOptions> _options;
		readonly IOptionsMonitor<FederationOptions> _federation;

		public CrawlPlanner(IJobQueue queue, IDomainBlocks blocks, IOptionsMonitor<StatisticsOptions> options, IOptionsMonitor<FederationOptions> federation = default)
		{
			_queue = queue;
			_blocks = blocks;
			_options = options;
			_federation = federation;
		}

		public JobKind Kind => JobKind.CrawlPlan;
		public int Concurrency => 1;
		public int MaxAttempts => 1;
		public int PerHostLimit => 1;

		public static Job For(DateTime hour) => new()
		{
			Kind = JobKind.CrawlPlan,
			Payload = hour.ToString("yyyy-MM-ddTHH", System.Globalization.CultureInfo.InvariantCulture),
			DedupeKey = "crawl-plan|" + hour.ToString("yyyy-MM-ddTHH", System.Globalization.CultureInfo.InvariantCulture),
			RunAt = hour
		};

		public async Task<JobOutcome> Handle(Job job, CancellationToken token)
		{
			var crawler = _options.CurrentValue.Crawler;
			if (!crawler.Enabled)
				return JobOutcome.Done;
			var now = DateTime.UtcNow;
			foreach (var seed in PeerHosts.Filter(crawler.Seeds ?? new(), host => !_blocks.IsSuspended(host), _federation?.CurrentValue.AllowPrivateNetworks == true))
				await DB.Default.Update<RemoteInstance>()
					.Match(i => i.Host == seed)
					.Modify(b => b.SetOnInsert(i => i.Seen, "crawled"))
					.Modify(b => b.SetOnInsert(i => i.FirstSeenAt, now))
					.Option(o => o.IsUpsert = true)
					.ExecuteAsync(token);

			var due = now.AddDays(-Math.Max(crawler.RevisitDays, 1));
			var perHour = Math.Clamp(crawler.HostsPerHour, 1, 600);
			var candidates = await DB.Default.Find<RemoteInstance>()
				.Match(i => (i.LastCrawledAt == null || i.LastCrawledAt < due) && (i.UnavailableUntil == null || i.UnavailableUntil < now))
				.Sort(i => i.LastCrawledAt, Order.Ascending)
				.Limit(perHour * 3)
				.ExecuteAsync(token);
			var week = InstanceDescriber.Week(now);
			var jobs = candidates.Where(i => _blocks.Find(i.Host) == default && PeerHosts.Normalise(i.Host) != default)
				.Take(perHour)
				.Select((instance, index) => new Job
				{
					Kind = JobKind.CrawlInstance,
					Payload = instance.Host,
					Host = instance.Host,
					DedupeKey = $"crawl|{instance.Host}|{week}",
					RunAt = now.AddSeconds(index * 3600.0 / perHour)
				});
			await _queue.EnqueueMany(jobs, token);
			return JobOutcome.Done;
		}
	}

	//one server: its robots.txt, a description if only crawling ever found it, then the servers its peers list names
	public class InstanceCrawler : IJobHandler
	{
		const int MaxPeers = 20_000;
		const int MaxRobotsBytes = 64 * 1024;

		readonly IFederationHttp _http;
		readonly InstanceDescriber _describer;
		readonly IDomainBlocks _blocks;
		readonly IOptionsMonitor<StatisticsOptions> _options;
		readonly IOptionsMonitor<AppConfiguration> _app;
		readonly IOptionsMonitor<FederationOptions> _federation;
		readonly IInteractionLedger _ledger;

		public InstanceCrawler(IFederationHttp http, InstanceDescriber describer, IDomainBlocks blocks, IOptionsMonitor<StatisticsOptions> options,
			IOptionsMonitor<AppConfiguration> app, IOptionsMonitor<FederationOptions> federation, IInteractionLedger ledger = default)
		{
			_http = http;
			_describer = describer;
			_blocks = blocks;
			_options = options;
			_app = app;
			_federation = federation;
			_ledger = ledger;
		}

		public JobKind Kind => JobKind.CrawlInstance;
		public int Concurrency => 1;
		public int MaxAttempts => 1;
		public int PerHostLimit => 1;

		protected virtual string Address(string url) => url;

		public async Task<JobOutcome> Handle(Job job, CancellationToken token)
		{
			var crawler = _options.CurrentValue.Crawler;
			var host = job.Payload;
			if (!crawler.Enabled || _blocks.Find(host) != default)
				return JobOutcome.Done;
			using var crawl = HttpScope.Crawl(Stargazer.UserAgent(_app.CurrentValue.BackendBaseAddress));

			RobotsTxt robots;
			using (HttpScope.For("robots"))
			{
				var (status, text) = await _http.GetText(Address($"https://{host}/robots.txt"), MaxRobotsBytes, token);
				robots = RobotsTxt.From(status, text, Stargazer.Token, "PrivaPub");
			}
			var instance = await DB.Default.Find<RemoteInstance>().Match(i => i.Host == host).ExecuteFirstAsync(token);
			var now = DateTime.UtcNow;
			if (!robots.Allows("/.well-known/nodeinfo") && !robots.Allows("/api/v1/instance/peers"))
			{
				await Visited(host, robotsDisallowed: true, peers: default, "robots.txt keeps us out", token);
				Record(host, Interactions.Refused, "robots-disallowed");
				return JobOutcome.Done;
			}

			var reachable = true;
			if (instance?.Seen != "touched")
				reachable = await _describer.Describe(host, crawled: true, robots.Allows, token);

			int? peersCount = default;
			if (robots.Allows("/api/v1/instance/peers"))
			{
				IReadOnlyList<string> peers;
				using (HttpScope.For("peers"))
					peers = await _http.GetStringArray(Address($"https://{host}/api/v1/instance/peers"), MaxPeers, token);
				peersCount = peers.Count;
				await Discover(peers, crawler, token);
			}
			await Visited(host, robotsDisallowed: false, peersCount, reachable ? default : "no NodeInfo", token);
			Record(host, reachable ? Interactions.Ok : Interactions.Failed, reachable ? default : "unreachable");
			return JobOutcome.Done;
		}

		async Task Discover(IReadOnlyList<string> peers, CrawlerOptions crawler, CancellationToken token)
		{
			var room = Math.Min(crawler.MaxNewHostsPerCrawl, crawler.MaxHosts - (int)await DB.Default.CountAsync<RemoteInstance>(i => i.Seen == "crawled", token));
			if (room <= 0)
				return;
			var candidates = PeerHosts.Filter(peers, host => _http.IsAllowed(new Uri($"https://{host}/")) && _blocks.Find(host) == default,
				_federation.CurrentValue.AllowPrivateNetworks).ToList();
			var now = DateTime.UtcNow;
			foreach (var batch in candidates.Chunk(500))
			{
				if (room <= 0)
					return;
				var known = (await DB.Default.Find<RemoteInstance, string>().Match(i => batch.Contains(i.Host)).Project(i => i.Host).ExecuteAsync(token)).ToHashSet();
				foreach (var host in batch.Where(h => !known.Contains(h)).Take(room))
				{
					await DB.Default.Update<RemoteInstance>()
						.Match(i => i.Host == host)
						.Modify(b => b.SetOnInsert(i => i.Seen, "crawled"))
						.Modify(b => b.SetOnInsert(i => i.FirstSeenAt, now))
						.Option(o => o.IsUpsert = true)
						.ExecuteAsync(token);
					room--;
				}
			}
		}

		static async Task Visited(string host, bool robotsDisallowed, int? peers, string error, CancellationToken token)
		{
			var update = DB.Default.Update<RemoteInstance>()
				.Match(i => i.Host == host)
				.Modify(i => i.LastCrawledAt, DateTime.UtcNow)
				.Modify(i => i.RobotsDisallowed, robotsDisallowed)
				.Modify(i => i.CrawlError, error)
				.Modify(b => b.SetOnInsert(i => i.Seen, "crawled"))
				.Option(o => o.IsUpsert = true);
			if (peers is { } count)
				update = update.Modify(i => i.PeersCount, count);
			await update.ExecuteAsync(token);
		}

		void Record(string host, string outcome, string reason) =>
			_ledger?.Record(new InteractionEvent { Channel = Interactions.Crawl, Host = host, Outcome = outcome, Reason = reason, Crawl = true });
	}
}