SocialPub/PrivaPub/Federation/Inbox/Handlers/AnnounceHandler.cs

using MongoDB.Driver;
using MongoDB.Entities;

using PrivaPub.Domain.Statuses;
using PrivaPub.Domain.Social;
using PrivaPub.Domain.Timelines;
using PrivaPub.Federation.Actors;
using PrivaPub.Federation.Objects;
using PrivaPub.Infrastructure.Ids;
using PrivaPub.Models.Federation;
using PrivaPub.Models.Post;
using PrivaPub.Models.Social;
using PrivaPub.Models.User;
using PrivaPub.StaticServices;

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

using static PrivaPub.Federation.Objects.ActivityJson;

using PostEntity = PrivaPub.Models.Post.Post;

namespace PrivaPub.Federation.Inbox.Handlers
{
	public class AnnounceHandler : IActivityHandler
	{
		readonly DbEntities _dbEntities;
		readonly ILocalActorService _localActors;
		readonly IRemotePosts _remotePosts;
		readonly IFanout _fanout;
		readonly IRemoteActorService _remoteActors;

		readonly IObjectRecords _records;
		readonly IQuoteService _quotes;
		readonly IServiceProvider _services;

		// the handlers a community's relayed votes go to (Like, Dislike, Undo); resolved when first needed, since this
		// handler is one of them; tests set it
		public IEnumerable<IActivityHandler> Relays { get; set; }

		public AnnounceHandler(DbEntities dbEntities, ILocalActorService localActors, IRemotePosts remotePosts, IFanout fanout, IRemoteActorService remoteActors,
			IObjectRecords records, IQuoteService quotes, IServiceProvider services = default)
		{
			_services = services;
			_records = records;
			_quotes = quotes;
			_remoteActors = remoteActors;
			_dbEntities = dbEntities;
			_localActors = localActors;
			_remotePosts = remotePosts;
			_fanout = fanout;
		}

		public string Type => "Announce";

		public async Task Handle(JsonNode activity, ForeignAvatar actor, CancellationToken token)
		{
			var inner = activity["object"];
			if (inner is JsonObject && Value(inner, "type") is "Create" or "Update" or "Delete" or "Like" or "Dislike" or "Undo" or "Add" or "Remove" or "Block" or "Lock")
			{
				await GroupActivity(inner, actor, token);
				return;
			}
			var objectUri = Id(inner);
			var announceId = Id(activity);
			if (objectUri == default || announceId == default)
			{
				Arrival.Drop("unparseable");
				return;
			}
			if (await _dbEntities.Posts.Match(p => p.ObjectURI == announceId).ExecuteAnyAsync(token))
			{
				Arrival.Drop("duplicate");
				return;
			}

			var isLocal = objectUri.StartsWith(_localActors.BaseAddress + "/", StringComparison.OrdinalIgnoreCase);
			var original = await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && !p.DeletedAt.HasValue && p.ReblogOfPostId == null).ExecuteFirstAsync(token);
			var followed = await _dbEntities.Followings.Match(f => f.TargetActorURI == actor.ActorURI && f.State == FollowState.Accepted).ExecuteAnyAsync(token);
			if (original is not { IsFederatedCopy: false } && !followed)
			{
				Arrival.Drop("not-followed");
				return;
			}
			if (original == default && !isLocal)
				original = await _remotePosts.StoreContext(objectUri, 0, token);
			if (original == default)
			{
				Arrival.Drop("fetch-failed");
				return;
			}
			Arrival.About(original.ObjectType ?? "Note", original.Visibility, original.CreationDate);
			if (original.Visibility is not (PostVisibility.Public or PostVisibility.Unlisted))
			{
				Arrival.Drop("not-public");
				return;
			}

			var to = Strings(activity["to"]);
			var cc = Strings(activity["cc"]);
			var published = DateTimeOffset.TryParse(Value(activity, "published"), CultureInfo.InvariantCulture, DateTimeStyles.AssumeUniversal, out var at)
				? at.UtcDateTime
				: DateTime.UtcNow;
			var reblog = new PostEntity
			{
				ID = PrivacyIds.Arrived(published),
				ObjectURI = announceId,
				ActivityURI = announceId,
				ActorURI = actor.ActorURI,
				AuthorAccountId = actor.ID,
				ReblogOfPostId = original.ID,
				Visibility = Addressing.Classify(to, cc, actor.FollowersURL),
				To = to,
				Cc = cc,
				IsFederatedCopy = true,
				CreationDate = published,
				UpdateDate = published
			};
			if (reblog.Visibility == PostVisibility.Direct)
			{
				Arrival.Drop("not-public");
				return;
			}
			try
			{
				await DB.Default.SaveAsync(reblog, token);
			}
			catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
			{
				Arrival.Drop("duplicate");
				return;
			}
			Arrival.Accept("stored");
			Arrival.About(local: original.IsFederatedCopy ? default : await _localActors.FindById(Models.Federation.LocalActorKind.Person, original.GroupUserId, token));

			await DB.Default.Update<PostEntity>().MatchID(original.ID).Modify(b => b.Inc(p => p.ReblogsCount, 1)).ExecuteAsync(token);
			if (!original.IsFederatedCopy)
				await Notifications.Add(original.GroupUserId, NotificationType.Reblog, actor.ID, actor.ActorURI, original.ID, token);
			await _fanout.Distribute(reblog, token);
		}

		async Task GroupActivity(JsonNode inner, ForeignAvatar group, CancellationToken token)
		{
			Arrival.About(Value(inner, "type"));
			if (group.AvatarType != AvatarType.Group
				|| !await _dbEntities.Followings.Match(f => f.TargetActorURI == group.ActorURI && f.State == FollowState.Accepted).ExecuteAnyAsync(token)
					&& !await OurPostInItsThread(inner, group, token))
			{
				Arrival.Drop("not-followed");
				return;
			}
			var objectUri = Id(inner["object"]);
			switch (Value(inner, "type"))
			{
				case "Create" when objectUri != default:
					if (await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri).ExecuteAnyAsync(token))
					{
						Arrival.Drop("duplicate");
						return;
					}
					var post = await _remotePosts.StoreContext(objectUri, 0, token);
					if (post == default)
					{
						Arrival.Drop("fetch-failed");
						return;
					}
					Arrival.Accept("stored");
					Arrival.About(visibility: post.Visibility);
					await DB.Default.Update<PostEntity>().MatchID(post.ID).Modify(p => p.AudienceURI, group.ActorURI).ExecuteAsync(token);
					post.AudienceURI = group.ActorURI;
					await _fanout.Distribute(post, token);
					break;
				case "Update" when objectUri != default:
					var stored = await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && p.AudienceURI == group.ActorURI && !p.DeletedAt.HasValue)
						.ExecuteFirstAsync(token);
					if (stored == default)
					{
						Arrival.Drop("unknown-object");
						return;
					}
					Arrival.About(visibility: stored.Visibility, created: stored.CreationDate);
					using (var fetched = await _remoteActors.FetchObject(objectUri, token))
					{
						var note = fetched == default ? default : NoteParser.Parse(System.Text.Json.Nodes.JsonNode.Parse(fetched.Root.GetRawText()));
						if (note == default || note.AttributedTo != stored.ActorURI)
						{
							Arrival.Drop(note == default ? "fetch-failed" : "misattributed");
							return;
						}
						Arrival.Accept(await RemoteEdits.Apply(stored, note, Id(inner), _localActors, _records, _quotes, token) ? "edit" : "refresh");
					}
					break;
				case "Delete" when objectUri != default:
					var deleted = await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && p.AudienceURI == group.ActorURI).ExecuteFirstAsync(token);
					if (deleted == default)
					{
						Arrival.Drop("unknown-object");
						return;
					}
					Arrival.About(visibility: deleted.Visibility, created: deleted.CreationDate);
					// a community on the post's own server speaks for it: PieFed keeps serving a thread its moderator removed.
					// One elsewhere is believed once the post's server says it is gone
					if (!Origin.Same(objectUri, group.ActorURI))
						using (var check = await _remoteActors.FetchObject(objectUri, token))
							if (check != default && Value(System.Text.Json.Nodes.JsonNode.Parse(check.Root.GetRawText()), "type") != "Tombstone")
							{
								Arrival.Drop("not-deleted");
								return;
							}
					Arrival.Accept("removed");
					await RemoteDeletes.Remove(deleted, objectUri, token);
					break;
				case "Lock" when objectUri != default:
					await Lock(objectUri, group, true, token);
					break;
				case "Undo" when Value(inner["object"], "type") == "Lock":
					await Lock(Id(inner["object"]?["object"]), group, false, token);
					break;
				case "Block" when Id(inner["target"]) == group.ActorURI:
					await Ban(inner, group, token);
					break;
				case "Undo" when Value(inner["object"], "type") == "Block" && Id(inner["object"]?["target"]) == group.ActorURI:
					if (await BlockHandler.Undo(inner["object"], Id(inner["object"]), group, _localActors, token))
						Arrival.Accept("unbanned");
					else
						Arrival.Drop("unknown-object");
					break;
				case "Like" or "Dislike" or "Undo":
					await Relayed(inner, group, token);
					break;
				// a moderator features a post in the community, or no longer: the community's own pins
				case "Add" or "Remove" when Id(inner["target"]) is { } target && target == group.FeaturedURL:
					var featured = (Relays ?? _services?.GetService(typeof(IEnumerable<IActivityHandler>)) as IEnumerable<IActivityHandler>)
						?.FirstOrDefault(h => h.Type == Value(inner, "type"));
					if (featured == default)
						Arrival.Drop("unsupported");
					else
						await featured.Handle(inner, group, token);
					break;
				default:
					Arrival.Drop("unsupported");
					break;
			}
		}

		// A vote (or its undoing) on one of our posts in a thread the community holds: the community relays it to the post's
		// server whether or not anyone here follows the community (Lemmy sends a vote to the community alone)
		async Task<bool> OurPostInItsThread(JsonNode inner, ForeignAvatar group, CancellationToken token)
		{
			var target = Value(inner, "type") switch
			{
				"Like" or "Dislike" => Id(inner["object"]),
				"Undo" when (inner["object"] as JsonObject)?["type"]?.GetValue<string>() is "Like" or "Dislike" => Id(inner["object"]?["object"]),
				_ => default
			};
			var post = target == default
				? default
				: await _dbEntities.Posts.Match(p => p.ObjectURI == target && !p.IsFederatedCopy && !p.DeletedAt.HasValue).ExecuteFirstAsync(token);
			for (var depth = 0; post != default && depth < RemotePosts.MaxDepth; depth++)
			{
				if (post.AudienceURI == group.ActorURI)
					return true;
				post = string.IsNullOrEmpty(post.AnsweringToPostId) ? default : await _dbEntities.Posts.MatchID(post.AnsweringToPostId).ExecuteFirstAsync(token);
			}
			return false;
		}

		// A community's moderators lock one of its posts, or unlock it: no more replies, ours included. The community vouches
		// for what is done to its own posts.
		async Task Lock(string objectUri, ForeignAvatar group, bool locked, CancellationToken token)
		{
			var post = objectUri == default
				? default
				: await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && p.AudienceURI == group.ActorURI).ExecuteFirstAsync(token);
			if (post == default)
			{
				Arrival.Drop("unknown-object");
				return;
			}
			Arrival.About(visibility: post.Visibility, created: post.CreationDate);
			await DB.Default.Update<PostEntity>().MatchID(post.ID).Modify(p => p.LockedAt, locked ? DateTime.UtcNow : null).ExecuteAsync(token);
			Arrival.Accept(locked ? "locked" : "unlocked");
		}

		// A community bans one of our personas (Lemmy's Block with the community as its target): kept as the community
		// blocking the persona, so its relationship says blocked_by and nothing of the persona's is posted there until the
		// Undo. A ban of someone from another server is no business of ours.
		async Task Ban(JsonNode inner, ForeignAvatar group, CancellationToken token)
		{
			if (Id(inner["object"]) is not { } uri || await _localActors.FindByUri(uri, token) is not { Kind: LocalActorKind.Person } persona)
			{
				Arrival.Drop("not-ours");
				return;
			}
			try
			{
				await DB.Default.SaveAsync(new BlockedBy { AvatarId = persona.Id, ActorURI = group.ActorURI, AccountId = group.ID, ActivityURI = Id(inner) }, token);
				Arrival.Accept("banned");
			}
			catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
			{
				Arrival.Drop("duplicate");
			}
		}

		// A vote, or its undoing, that a community relays (Lemmy, PieFed and Mbin send them so). The community vouches for
		// what accounts on its own server do and for what is done to its own posts, as Lemmy trusts it (refetching every vote
		// would not scale); anything else is believed only once fetched from its own origin. It is then handled as if its
		// actor had delivered it.
		async Task Relayed(JsonNode inner, ForeignAvatar group, CancellationToken token)
		{
			var actorUri = Id(inner["actor"]);
			var type = Value(inner, "type");
			var handlers = Relays ?? _services?.GetService(typeof(IEnumerable<IActivityHandler>)) as IEnumerable<IActivityHandler>;
			var handler = handlers?.FirstOrDefault(h => h.Type == type);
			if (actorUri == default || handler == default)
			{
				Arrival.Drop("unparseable");
				return;
			}
			var activity = inner;
			var target = Id(Value(inner, "type") == "Undo" ? (inner["object"] as JsonObject)?["object"] : inner["object"]);
			var ownPost = target != default
				&& await _dbEntities.Posts.Match(p => p.ObjectURI == target && p.AudienceURI == group.ActorURI).ExecuteAnyAsync(token);
			if (!Origin.Same(actorUri, group.ActorURI) && !ownPost)
			{
				var id = Id(inner);
				if (id == default || !Origin.Same(id, actorUri))
				{
					Arrival.Drop("misattributed");
					return;
				}
				using var fetched = await _remoteActors.FetchObject(id, token);
				activity = fetched == default ? default : JsonNode.Parse(fetched.Root.GetRawText());
				if (activity == default || Value(activity, "type") != type || Id(activity["actor"]) != actorUri)
				{
					Arrival.Drop(activity == default ? "fetch-failed" : "misattributed");
					return;
				}
			}
			var actor = await _remoteActors.GetActor(actorUri, refresh: false, token);
			if (actor == default)
			{
				Arrival.Drop("unknown-actor");
				return;
			}
			await handler.Handle(activity, actor, token);
		}

		static List<string> Strings(JsonNode node) => node switch
		{
			JsonArray array => array.Select(Id).Where(id => id != default).ToList(),
			JsonNode single when Id(single) is { } id => new List<string> { id },
			_ => new List<string>()
		};
	}
}