decePubClient/Services/StreamingService.cs

using System.Net.WebSockets;
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;

using decePubClient.Helpers;
using decePubClient.Models;
using decePubClient.Models.Mastodon;

namespace decePubClient.Services
{
	/// <summary>One stream of PrivaPub's streaming API: its name and, for a hashtag or a list, which one.</summary>
	public sealed record StreamSubscription(string Stream, string Tag = default, string List = default);

	/// <summary>What a stream told: the stream it came on, the event (update, notification, status.update, delete) and
	/// what it is about (a Status or a Notification for those, the deleted id for delete).</summary>
	public sealed record StreamMessage(string Event, Status Status = default, Notification Notification = default, string DeletedId = default);

	/// <summary>What the user stream told of the administrators' announcements: one new or changed (announcement), one
	/// gone (announcement.delete, its id), or a reaction's count now (announcement.reaction).</summary>
	public sealed record AnnouncementMessage(string Event, Announcement Announcement = default, string DeletedId = default,
		AnnouncementReactionEvent Reaction = default);

	/// <summary>
	/// PrivaPub's streaming API (/api/v1/streaming) as the current persona: one WebSocket, its token given as the
	/// socket's protocol (never in a URL), subscribed to the streams the page asks for. What comes is raised as
	/// StreamMessages; when the socket drops it is opened again after a pause that grows to a minute, and IsOpen tells the
	/// page to poll meanwhile. Nothing throws.
	/// The announcements come on the user stream, which HearAnnouncements keeps subscribed whatever feed is followed; what
	/// else it tells is raised only while the feed followed is the user stream itself.
	/// </summary>
	public sealed class StreamingService(TokenAuthStateProvider authStateProvider, IHttpService httpService, AppConfiguration configuration,
		ILoggingService logger) : IAsyncDisposable
	{
		const string UserStream = "user";

		ClientWebSocket socket;
		CancellationTokenSource running;
		StreamSubscription current;
		bool announcements;

		public bool IsOpen => socket?.State == WebSocketState.Open && current is not null;

		/// <summary>Raised for each message of the stream subscribed now.</summary>
		public event Func<StreamMessage, Task> Received;

		/// <summary>Raised for each announcement event, while HearAnnouncements is on.</summary>
		public event Func<AnnouncementMessage, Task> Announced;

		Uri Endpoint
		{
			get
			{
				var root = new UriBuilder(configuration.ApiBaseAddress) { Path = "/api/v1/streaming" };
				root.Scheme = root.Scheme == "https" ? "wss" : "ws";
				return root.Uri;
			}
		}

		/// <summary>Listens to one stream (the feed in view), or to none; a change subscribes the open socket anew.</summary>
		public async Task Follow(StreamSubscription subscription, CancellationToken cancellationToken = default)
		{
			if (subscription == current && (running is not null || subscription is null && !announcements))
				return;
			var previous = current;
			current = subscription;
			if (socket?.State == WebSocketState.Open)
			{
				// the user stream stays while the announcements are heard on it
				if (previous is not null && !(announcements && previous.Stream == UserStream))
					await Say("unsubscribe", previous, cancellationToken);
				if (subscription is not null)
					await Say("subscribe", subscription, cancellationToken);
				return;
			}
			Start();
		}

		/// <summary>Hears the announcements (the user stream) or stops hearing them; the socket opens for them alone when no
		/// feed is followed.</summary>
		public async Task HearAnnouncements(bool hear, CancellationToken cancellationToken = default)
		{
			if (hear == announcements)
			{
				if (hear)
					Start();
				return;
			}
			announcements = hear;
			if (socket?.State == WebSocketState.Open)
			{
				if (current?.Stream != UserStream)
					await Say(hear ? "subscribe" : "unsubscribe", new StreamSubscription(UserStream), cancellationToken);
				return;
			}
			if (hear)
				Start();
		}

		void Start()
		{
			if (running is not null || current is null && !announcements)
				return;
			running = new();
			_ = Run(running.Token);
		}

		async Task Say(string type, StreamSubscription subscription, CancellationToken cancellationToken)
		{
			var message = new JsonObject { ["type"] = type, ["stream"] = subscription.Stream };
			if (subscription.Tag is not null)
				message["tag"] = subscription.Tag;
			if (subscription.List is not null)
				message["list"] = subscription.List;
			try
			{
				await socket.SendAsync(Encoding.UTF8.GetBytes(message.ToJsonString()), WebSocketMessageType.Text, true, cancellationToken);
			}
			catch (Exception ex) when (ex is WebSocketException or OperationCanceledException or InvalidOperationException)
			{
			}
		}

		// open, subscribe, read until the socket drops, and open again after a pause
		async Task Run(CancellationToken cancellationToken)
		{
			var pause = TimeSpan.FromSeconds(2);
			while (!cancellationToken.IsCancellationRequested)
			{
				try
				{
					await Connect(cancellationToken);
					pause = TimeSpan.FromSeconds(2);
					await Read(cancellationToken);
				}
				catch (OperationCanceledException)
				{
					return;
				}
				catch (Exception ex) when (ex is WebSocketException or JsonException or InvalidOperationException)
				{
					await logger.ProcessWarning(ex.Message, $"{nameof(StreamingService)}.{nameof(Run)}", cancellationToken);
				}
				socket?.Dispose();
				socket = null;
				try
				{
					await Task.Delay(pause, cancellationToken);
				}
				catch (OperationCanceledException)
				{
					return;
				}
				pause = TimeSpan.FromSeconds(Math.Min(60, pause.TotalSeconds * 2));
			}
		}

		async Task Connect(CancellationToken cancellationToken)
		{
			var token = await authStateProvider.GetPersonaToken(cancellationToken);
			if (token is null)
			{
				// any persona request exchanges the token first
				using var _ = await httpService.Get(APIs.Mastodon.GET_VerifyCredentials, cancellationToken: cancellationToken);
				token = await authStateProvider.GetPersonaToken(cancellationToken);
			}
			if (token is null)
				throw new InvalidOperationException("no persona token");
			socket = new ClientWebSocket();
			socket.Options.AddSubProtocol(token);
			await socket.ConnectAsync(Endpoint, cancellationToken);
			if (current is not null)
				await Say("subscribe", current, cancellationToken);
			if (announcements && current?.Stream != UserStream)
				await Say("subscribe", new StreamSubscription(UserStream), cancellationToken);
		}

		async Task Read(CancellationToken cancellationToken)
		{
			var buffer = new byte[64 * 1024];
			while (socket.State == WebSocketState.Open)
			{
				var text = new StringBuilder();
				WebSocketReceiveResult received;
				do
				{
					received = await socket.ReceiveAsync(buffer, cancellationToken);
					if (received.MessageType == WebSocketMessageType.Close)
						return;
					text.Append(Encoding.UTF8.GetString(buffer, 0, received.Count));
				}
				while (!received.EndOfMessage);
				if (JsonNode.Parse(text.ToString()) is not JsonObject message || message["event"]?.GetValue<string>() is not { } name
					|| message["payload"]?.GetValue<string>() is not { } payload)
					continue;
				if (name.StartsWith("announcement", StringComparison.Ordinal))
				{
					var announced = name switch
					{
						"announcement" => new AnnouncementMessage(name, Announcement: JsonSerializer.Deserialize<Announcement>(payload, SUtility.MastodonSerializer)),
						"announcement.delete" => new AnnouncementMessage(name, DeletedId: payload),
						"announcement.reaction" => new AnnouncementMessage(name,
							Reaction: JsonSerializer.Deserialize<AnnouncementReactionEvent>(payload, SUtility.MastodonSerializer)),
						_ => null
					};
					if (announced is not null && announcements && Announced is { } listeners)
						await listeners(announced);
					continue;
				}
				// the user stream held for the announcements tells the home feed's posts and notifications too: they are the
				// feed's only while it is the one followed
				if ((message["stream"] as JsonArray)?.FirstOrDefault()?.GetValue<string>() == UserStream && current?.Stream != UserStream)
					continue;
				var told = name switch
				{
					"update" or "status.update" => new StreamMessage(name, Status: JsonSerializer.Deserialize<Status>(payload, SUtility.MastodonSerializer)),
					"notification" => new StreamMessage(name, Notification: JsonSerializer.Deserialize<Notification>(payload, SUtility.MastodonSerializer)),
					"delete" => new StreamMessage(name, DeletedId: payload),
					_ => null
				};
				if (told is not null && Received is { } handlers)
					await handlers(told);
			}
		}

		/// <summary>Closes the socket; the page polls from then on. The announcements stay asked for: the next Follow opens
		/// the socket for them too.</summary>
		public async Task Stop()
		{
			current = null;
			if (running is not null)
			{
				await running.CancelAsync();
				running.Dispose();
				running = null;
			}
			try
			{
				if (socket?.State == WebSocketState.Open)
					await socket.CloseAsync(WebSocketCloseStatus.NormalClosure, default, CancellationToken.None);
			}
			catch (WebSocketException)
			{
			}
			socket?.Dispose();
			socket = null;
		}

		public async ValueTask DisposeAsync() => await Stop();
	}
}