diff --git a/CLAUDE.md b/CLAUDE.md
index 5550b41..0c58b47 100644
--- a/CLAUDE.md
+++ b/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
- the page is visible (PrivaPub has no streaming) and wait behind a "new" button.
+ `GlobePanel`. Older pages load as the end comes near (`wwwroot/js/feed.js`). Newer ones come through PrivaPub's
+ 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
diff --git a/Pages/Index.razor b/Pages/Index.razor
index 11fef0a..6849eef 100644
--- a/Pages/Index.razor
+++ b/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
- 45 seconds while the page is visible and shown on request, because PrivaPub has no streaming. *@
+ default on phones. Pages of older posts load as the end comes near (wwwroot/js/feed.js). Newer ones come through
+ 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. *@
@@ -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(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("isVisible", cancellationToken))
continue;
@@ -322,6 +329,45 @@
}
}
+ /// The stream that tells this feed's new posts, or null for a feed PrivaPub does not stream.
+ 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
+ };
+
+ /// 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.
+ 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();
diff --git a/Program.cs b/Program.cs
index 7cb0604..a6c64a1 100644
--- a/Program.cs
+++ b/Program.cs
@@ -45,6 +45,7 @@ builder.Services.AddOptions()
.AddScoped()
.AddScoped()
.AddScoped()
+ .AddScoped()
.AddScoped()
.AddScoped()
.AddScoped()
diff --git a/Resources/AllStrings.it.resx b/Resources/AllStrings.it.resx
index 9ff3c80..f988115 100644
--- a/Resources/AllStrings.it.resx
+++ b/Resources/AllStrings.it.resx
@@ -1190,4 +1190,10 @@
Non sarĂ pubblicato.
+
+ flusso
+
+
+ chiuso, interrogo
+
diff --git a/Resources/AllStrings.resx b/Resources/AllStrings.resx
index 608bdac..dc250ce 100644
--- a/Resources/AllStrings.resx
+++ b/Resources/AllStrings.resx
@@ -1190,4 +1190,10 @@
It will not be published.
+
+ stream
+
+
+ closed, polling
+
diff --git a/Services/StreamingService.cs b/Services/StreamingService.cs
new file mode 100644
index 0000000..7f4325d
--- /dev/null
+++ b/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
+{
+ /// One stream of PrivaPub's streaming API: its name and, for a hashtag or a list, which one.
+ public sealed record StreamSubscription(string Stream, string Tag = default, string List = default);
+
+ /// 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).
+ public sealed record StreamMessage(string Event, Status Status = default, Notification Notification = default, string DeletedId = default);
+
+ ///
+ /// 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.
+ ///
+ 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;
+
+ /// Raised for each message of the stream subscribed now.
+ public event Func Received;
+
+ Uri Endpoint
+ {
+ get
+ {
+ var root = new UriBuilder(configuration.ApiBaseAddress) { Path = "/api/v1/streaming" };
+ root.Scheme = root.Scheme == "https" ? "wss" : "ws";
+ return root.Uri;
+ }
+ }
+
+ /// Listens to one stream (the feed in view), or to none; a change subscribes the open socket anew.
+ 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() is not { } name
+ || message["payload"]?.GetValue() is not { } payload)
+ continue;
+ var told = name switch
+ {
+ "update" or "status.update" => new StreamMessage(name, Status: JsonSerializer.Deserialize(payload, SUtility.MastodonSerializer)),
+ "notification" => new StreamMessage(name, Notification: JsonSerializer.Deserialize(payload, SUtility.MastodonSerializer)),
+ "delete" => new StreamMessage(name, DeletedId: payload),
+ _ => null
+ };
+ if (told is not null && Received is { } handlers)
+ await handlers(told);
+ }
+ }
+
+ /// Closes the socket; the page polls from then on.
+ 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();
+ }
+}
diff --git a/tests/e2e/Actions/StreamingTests.cs b/tests/e2e/Actions/StreamingTests.cs
new file mode 100644
index 0000000..573883e
--- /dev/null
+++ b/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()}']");
+ await Assertions.Expect(card).ToContainTextAsync(text, new() { Timeout = 5000 });
+
+ await PersonaApi.Call(persona, $"/api/v1/statuses/{post["id"]!.GetValue()}", 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()}", HttpMethod.Delete);
+ }
+ }
+ }
+}