SocialPub/PrivaPub/Federation/Objects/ObjectRecords.cs

using MongoDB.Driver;
using MongoDB.Entities;

using PrivaPub.Federation.Inbox;
using PrivaPub.Infrastructure.Jobs;
using PrivaPub.Models.Federation;
using PrivaPub.Models.Jobs;

using System.Globalization;
using System.Security.Cryptography;
using System.Text;
using System.Text.Json.Nodes;

using PostEntity = PrivaPub.Models.Post.Post;

namespace PrivaPub.Federation.Objects
{
	public interface IObjectRecords
	{
		Task Record(NoteDocument note, PostEntity post, ObjectPath path, bool refetched, CancellationToken token);
		Task Revise(NoteDocument note, string activityId, CancellationToken token);
	}

	public class ObjectRecords : IObjectRecords
	{
		public const int MaxRawBytes = 256 * 1024;
		const int MaxRevisions = 10;
		static readonly TimeSpan DescriptionAge = TimeSpan.FromDays(7);

		readonly IJobQueue _queue;

		public ObjectRecords(IJobQueue queue) => _queue = queue;

		public async Task Record(NoteDocument note, PostEntity post, ObjectPath path, bool refetched, CancellationToken token)
		{
			var arrival = Arrival.Current;
			var delivered = path == ObjectPath.Delivered;
			var raw = Capture(note.Raw);
			var host = new Uri(note.Id).Host.ToLowerInvariant();
			var activityContext = !delivered || note.Raw?.ContainsKey("@context") == true || arrival?.Context?.Length > 16 * 1024 ? default : arrival?.Context;
			var record = new ObjectRecord
			{
				ObjectURI = note.Id,
				PostId = post.ID,
				ObjectType = note.Type,
				Host = host,
				Raw = raw.Text,
				RawHash = raw.Hash,
				RawBytes = raw.Bytes,
				RawTruncated = raw.Truncated,
				ActivityContext = activityContext,
				Path = path,
				Refetched = refetched,
				ActivityId = arrival?.ActivityId,
				ActivityType = arrival?.ActivityType,
				ActivityActorURI = arrival?.ActorURI,
				Inbox = delivered ? arrival?.Inbox : default,
				SignatureScheme = !delivered || arrival?.KeyId == default ? default : "draft-cavage",
				KeyId = delivered ? arrival?.KeyId : default,
				Algorithm = delivered ? arrival?.Algorithm : default,
				SignedHeaders = delivered ? arrival?.SignedHeaders?.ToList() ?? new() : new(),
				ReceivedAt = delivered ? arrival?.ReceivedAt ?? DateTime.UtcNow : DateTime.UtcNow,
				Published = note.Published,
				Updated = note.Updated,
				Extensions = ObjectFeatures.Detect(note.Raw),
				ContextNamespaces = ObjectFeatures.Namespaces(note.Raw, activityContext)
			};
			try
			{
				await DB.Default.SaveAsync(record, token);
			}
			catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
			{
				return;
			}
			await Describe(host, token);
		}

		public async Task Revise(NoteDocument note, string activityId, CancellationToken token)
		{
			var raw = Capture(note.Raw);
			var revision = new ObjectRevision
			{
				Raw = raw.Text,
				RawHash = raw.Hash,
				RawTruncated = raw.Truncated,
				ActivityId = activityId,
				Updated = note.Updated
			};
			await DB.Default.Update<ObjectRecord>()
				.Match(r => r.ObjectURI == note.Id)
				.Modify(b => b.PushEach(r => r.Revisions, new[] { revision }, -MaxRevisions))
				.Modify(r => r.Updated, note.Updated)
				.ExecuteAsync(token);
		}

		async Task Describe(string host, CancellationToken token)
		{
			var known = await DB.Default.Find<RemoteInstance>().Match(i => i.Host == host).ExecuteFirstAsync(token);
			if (known?.DescribedAt > DateTime.UtcNow - DescriptionAge)
				return;
			await _queue.Enqueue(JobKind.DescribeInstance, host, host, InstanceDescriber.DedupeKey(host, DateTime.UtcNow), token);
		}

		public static (string Text, string Hash, int Bytes, bool Truncated) Capture(JsonNode raw)
		{
			if (raw == default)
				return default;
			var text = raw.ToJsonString();
			var bytes = Encoding.UTF8.GetBytes(text);
			var hash = Convert.ToHexStringLower(SHA256.HashData(bytes));
			return bytes.Length > MaxRawBytes ? (default, hash, bytes.Length, true) : (text, hash, bytes.Length, false);
		}
	}
}