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() .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().Match(i => i.Host == host).ExecuteFirstAsync(token); if (known?.DescribedAt > DateTime.UtcNow - DescriptionAge) return; var week = ISOWeek.GetYear(DateTime.UtcNow) + "-" + ISOWeek.GetWeekOfYear(DateTime.UtcNow); await _queue.Enqueue(JobKind.DescribeInstance, host, host, $"describe|{host}|{week}", 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); } } }