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>
	/// 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.
	/// </summary>
	public sealed class StreamingService(TokenAuthStateProvider authStateProvider, IHttpService httpService, AppConfiguration configuration,
		ILoggingService logger) : IAsyncDisposable
	{
		ClientWebSocket socket;
		CancellationTokenSource running;
		StreamSubscription current;

		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;

		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))
				return;
			var previous = current;
			current = subscription;
			if (socket?.State == WebSocketState.Open)
			{
				if (previous is not null)
					await Say("unsubscribe", previous, cancellationToken);
				if (subscription is not null)
					await Say("subscribe", subscription, cancellationToken);
				return;
			}
			if (subscription is not null && running is null)
			{
				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);
		}

		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;
				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.</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();
	}
}