The feed hears new posts as they come

decePub opens PrivaPub's streaming API (one WebSocket for the persona, its
token given as the socket's protocol, never in a URL) and subscribes to the
stream of the feed in view: home, a list, local, federated, a hashtag or
notifications. New posts wait behind the "new" button as polled ones did;
a deletion or an edit applies at once. Polling stays for the feeds PrivaPub
does not stream and for any feed while the socket is down, which opens again
after a pause growing to a minute. The e2e test sees a post made elsewhere
in seconds and its deletion take the card away.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LsXgEaXee4GCU1hwYgPJXw

CLAUDE.md

@@ -242,8 +242,10 @@ supplies:
   are read on the first hover and kept for the session. PrivaPub's own host is the signed-in persona's account host,
   since the API may be reached under another name.
 - **The feed page** (`Pages/Index.razor`): `FeedTabs`, `Content` (the card, which does its own actions), `Composer`,
-  `GlobePanel`. Older pages load as the end comes near (`wwwroot/js/feed.js`); newer ones are polled every 45 s while
+  `GlobePanel`. Older pages load as the end comes near (`wwwroot/js/feed.js`). Newer ones come through PrivaPub's
-  the page is visible (PrivaPub has no streaming) and wait behind a "new" button.
+  streaming API (`Services/StreamingService.cs`: one WebSocket, the persona's token as its protocol, never in a URL)
+  for home, lists, local, federated, hashtags and notifications, and wait behind a "new" button; a deletion or an edit
+  it tells applies at once. Other feeds, and any feed while the stream is down, are polled every 45 s while visible.
 - **The globe** (`wwwroot/js/globe.js`, an ES module imported by `GlobePanel`) runs on globe.gl, vendored in
   `wwwroot/vendor/globe/` with its textures and `countries.json` (`SOURCES.md` has versions, licences and hashes; the
   service worker never precaches it). It looks at the spherical mean of the located posts in view, pulses and draws an

Pages/Index.razor

@@ -7,8 +7,10 @@
 
 @* The feed page, the middle and right of three columns (NavMenu is the left): the feed with a tab per kind of feed,
    and the globe, which follows the posts in view (GlobePanel). Below lg the globe is a banner above the feed, closed by
-   default on phones. Pages of older posts load as the end comes near (wwwroot/js/feed.js); newer ones are polled every
+   default on phones. Pages of older posts load as the end comes near (wwwroot/js/feed.js). Newer ones come through
-   45 seconds while the page is visible and shown on request, because PrivaPub has no streaming. *@
+   PrivaPub's streaming API (StreamingService) for the feeds it streams, and are polled every 45 seconds while the page
+   is visible otherwise, or while the stream is down; either way they wait behind the "new" button. A deletion or an edit
+   the stream tells is applied at once. *@
 <div class="flex flex-col lg:grid lg:grid-cols-[minmax(0,1fr)_minmax(320px,42%)] gap-3 md:gap-4 w-full h-full">
 
 	<aside class="@(Filter.ShowGlobe ? "block" : "hidden") lg:block lg:order-last relative h-56 sm:h-64 lg:h-full flex-none min-h-0 rounded-xl neo-raised neo-xs overflow-hidden">
@@ -104,6 +106,7 @@
 	[Inject] TimelineService TimelineService { get; set; }
 	[Inject] GroupService GroupService { get; set; }
 	[Inject] ListService ListService { get; set; }
+	[Inject] StreamingService Streaming { get; set; }
 	[Inject] InstanceService InstanceService { get; set; }
 	[Inject] RequestStats RequestStats { get; set; }
 
@@ -133,6 +136,7 @@
 
 	protected override async Task OnInitializedAsync()
 	{
+		Streaming.Received += OnStream;
 		Filter = await LocalStorage.GetItemAsync<FeedFilter>(nameof(FeedFilter)) ?? new() { ShowGlobe = !CascadingState.Status.IsMobileMedia() };
 		if (Filter.TimelineType is TimelineType.Tag && Filter.Tag is not { Length: > 0 })
 			Filter.TimelineType = TimelineType.Home;
@@ -205,6 +209,8 @@
 
 		if (CascadingState.PersonaId != loadedPersonaId)
 		{
+			// a stream is the persona's: another one opens its own
+			await Streaming.Stop();
 			groups = await GroupService.List(CascadingState.PersonaId, cancellationToken);
 			lists = await ListService.All(cancellationToken);
 			loadedPersonaId = CascadingState.PersonaId;
@@ -238,6 +244,7 @@
 		maxId = page.MaxId;
 		minId = page.MinId;
 		lastPollAt = DateTime.UtcNow;
+		await Streaming.Follow(StreamOf(Filter), cancellationToken);
 		StateHasChanged();
 		await Locate(posts, cancellationToken);
 		if (feedModule is not null)
@@ -290,7 +297,7 @@
 		{
 			while (await pollTimer.WaitForNextTickAsync(cancellationToken))
 			{
-				if (IsLoading || CascadingState.IsOffline || minId is null || Filter.TimelineType is TimelineType.Nearby or TimelineType.Direct
+				if (IsLoading || CascadingState.IsOffline || minId is null || Streaming.IsOpen || Filter.TimelineType is TimelineType.Nearby or TimelineType.Direct
 					|| Filter.TimelineType is TimelineType.Groups && groups.FirstOrDefault(g => g.Id == Filter.GroupId)?.IsCommunity != true
 					|| !await feedModule.InvokeAsync<bool>("isVisible", cancellationToken))
 					continue;
@@ -322,6 +329,45 @@
 		}
 	}
 
+	/// <summary>The stream that tells this feed's new posts, or null for a feed PrivaPub does not stream.</summary>
+	static StreamSubscription StreamOf(FeedFilter filter) => filter.TimelineType switch
+	{
+		TimelineType.Home => new("user"),
+		TimelineType.Local => new("public:local"),
+		TimelineType.Federation => new("public:remote"),
+		TimelineType.Notifications => new("user:notification"),
+		TimelineType.Tag when filter.Tag is { Length: > 0 } tag => new("hashtag", Tag: tag),
+		TimelineType.Lists when filter.ListId is { Length: > 0 } list => new("list", List: list),
+		_ => null
+	};
+
+	/// <summary>What the stream tells: a new post or notification waits behind the "new" button; a deletion or an edit is
+	/// applied where the post is.</summary>
+	async Task OnStream(StreamMessage message)
+	{
+		switch (message.Event)
+		{
+			case "update" when message.Status is { } status && Filter.TimelineType is not TimelineType.Notifications:
+			case "notification" when message.Notification is not null && Filter.TimelineType is TimelineType.Notifications:
+				var post = message.Status is { } s ? FeedPost.Of(s) : FeedPost.Of(message.Notification);
+				if (posts.Concat(pending).Any(p => p.Key == post.Key))
+					return;
+				pending.Insert(0, post);
+				break;
+			case "delete" when message.DeletedId is { } id:
+				posts.RemoveAll(p => p.Key == id || p.Status?.Id == id);
+				pending.RemoveAll(p => p.Key == id || p.Status?.Id == id);
+				break;
+			case "status.update" when message.Status is { } edited:
+				foreach (var shown in posts.Concat(pending).Where(p => p.Status?.Id == edited.Id))
+					shown.Status = edited;
+				break;
+			default:
+				return;
+		}
+		await InvokeAsync(StateHasChanged);
+	}
+
 	async Task ShowPending()
 	{
 		var shown = pending;
@@ -400,6 +446,7 @@
 		yield return NerdRow.Of("max_id", maxId ?? (posts.Count > 0 ? Localizer["end"] : null));
 		yield return NerdRow.Of(Localizer["first page"], lastLoadTook is { } took ? NerdRow.Duration(took) : null);
 		yield return NerdRow.Of(Localizer["polled"], lastPollAt is { } polled ? NerdRow.Duration(DateTime.UtcNow - polled) + " " + Localizer["ago"] : null);
+		yield return NerdRow.Of(Localizer["stream"], Streaming.IsOpen ? Localizer["open"] : Localizer["closed, polling"]);
 		yield return NerdRow.Of(Localizer["poll every"], NerdRow.Duration(PollInterval));
 		yield return NerdRow.Of(Localizer["placed"], posts.Count > 0 ? $"{posts.Count(p => p.Location is not null)}/{posts.Count}" : null);
 		if (RequestStats.Latest(1).FirstOrDefault() is { } last)
@@ -412,6 +459,8 @@
 
 	public override async ValueTask DisposeAsync()
 	{
+		Streaming.Received -= OnStream;
+		await Streaming.Stop();
 		await pageCancellation.CancelAsync();
 		pageCancellation.Dispose();
 		loadCancellation?.Cancel();

Program.cs

@@ -45,6 +45,7 @@ builder.Services.AddOptions()
 	.AddScoped<ListService>()
 	.AddScoped<FilterService>()
 	.AddScoped<DiscoveryService>()
+	.AddScoped<StreamingService>()
 	.AddScoped<PublicNoticeService>()
 	.AddScoped<MediaService>()
 	.AddScoped<GroupService>()

Resources/AllStrings.it.resx

@@ -1190,4 +1190,10 @@
   <data name="It will not be published." xml:space="preserve">
     <value>Non sarà pubblicato.</value>
   </data>
+  <data name="stream" xml:space="preserve">
+    <value>flusso</value>
+  </data>
+  <data name="closed, polling" xml:space="preserve">
+    <value>chiuso, interrogo</value>
+  </data>
 </root>

Resources/AllStrings.resx

@@ -1190,4 +1190,10 @@
   <data name="It will not be published." xml:space="preserve">
     <value>It will not be published.</value>
   </data>
+  <data name="stream" xml:space="preserve">
+    <value>stream</value>
+  </data>
+  <data name="closed, polling" xml:space="preserve">
+    <value>closed, polling</value>
+  </data>
 </root>

Services/StreamingService.cs

@@ -0,0 +1,191 @@
+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();
+	}
+}

tests/e2e/Actions/StreamingTests.cs

@@ -0,0 +1,52 @@
+using System.Text.Json.Nodes;
+
+using decePubClient.E2E.Support;
+
+using Microsoft.Playwright;
+
+namespace decePubClient.E2E.Actions
+{
+	// The home feed hears of a new post and of its deletion through PrivaPub's streaming API, in seconds, where polling
+	// would take up to 45
+	public sealed class StreamingTests(Browser browser)
+	{
+		[Fact]
+		public async Task A_post_made_elsewhere_waits_behind_the_new_button_and_its_deletion_takes_it_away()
+		{
+			var persona = Pasture.Personas().First();
+			var text = $"streamed {Guid.NewGuid():N}"[..21];
+			await using var session = await SignIn.As(browser, persona);
+			var page = session.Page;
+			JsonNode post = null;
+			try
+			{
+				await Results.Cell("client.render.stream", "privapub", persona.UserName, async () =>
+				{
+					await page.GotoAsync($"{Pasture.Site}/");
+					await page.EvaluateAsync("() => localStorage.removeItem('FeedFilter')");
+					await page.GotoAsync($"{Pasture.Site}/");
+					await page.Locator("article[data-uri], [data-feed-empty]").First.WaitForAsync(new() { Timeout = 30000 });
+					// the stream opens once the first page is shown
+					await Task.Delay(1500);
+
+					post = await PersonaApi.Call(persona, "/api/v1/statuses", HttpMethod.Post, new JsonObject { ["status"] = text, ["visibility"] = "unlisted" });
+					var fresh = page.GetByRole(AriaRole.Button, new() { Name = "1 new" });
+					await fresh.WaitForAsync(new() { Timeout = 10000 });
+					await fresh.ClickAsync();
+					var card = page.Locator($"article[data-uri='{post!["uri"]!.GetValue<string>()}']");
+					await Assertions.Expect(card).ToContainTextAsync(text, new() { Timeout = 5000 });
+
+					await PersonaApi.Call(persona, $"/api/v1/statuses/{post["id"]!.GetValue<string>()}", HttpMethod.Delete);
+					post = null;
+					await Assertions.Expect(card).ToHaveCountAsync(0, new() { Timeout = 10000 });
+					await session.AssertGuards();
+				}, page);
+			}
+			finally
+			{
+				if (post is not null)
+					await PersonaApi.Call(persona, $"/api/v1/statuses/{post["id"]!.GetValue<string>()}", HttpMethod.Delete);
+			}
+		}
+	}
+}