Merge branch 'stream-marker-updates' into 'develop'

Stream marker updates

See merge request pleroma/pleroma!4354

lib/pleroma/notification.ex

@@ -282,10 +282,15 @@ defmodule Pleroma.Notification do
         select: n.id
       )
 
-    Multi.new()
-    |> Multi.update_all(:ids, query, set: [seen: true, updated_at: NaiveDateTime.utc_now()])
-    |> Marker.multi_set_last_read_id(user, "notifications")
-    |> Repo.transaction()
+    {:ok, %{marker: marker}} =
+      Multi.new()
+      |> Multi.update_all(:ids, query, set: [seen: true, updated_at: NaiveDateTime.utc_now()])
+      |> Marker.multi_set_last_read_id(user, "notifications")
+      |> Repo.transaction()
+
+    Streamer.stream(["user", "user:notification"], marker)
+
+    {:ok, %{marker: marker}}
   end
 
   @spec read_one(User.t(), String.t()) ::

lib/pleroma/web/streamer.ex

@@ -10,6 +10,7 @@ defmodule Pleroma.Web.Streamer do
   alias Pleroma.Chat.MessageReference
   alias Pleroma.Config
   alias Pleroma.Conversation.Participation
+  alias Pleroma.Marker
   alias Pleroma.Notification
   alias Pleroma.Object
   alias Pleroma.User
@@ -321,6 +322,16 @@ defmodule Pleroma.Web.Streamer do
     end)
   end
 
+  defp do_stream(topic, %Marker{} = marker) do
+    Registry.dispatch(@registry, "#{topic}:#{marker.user_id}", fn list ->
+      Enum.each(list, fn {pid, _auth} ->
+        text = StreamerView.render("marker.json", marker)
+
+        send(pid, {:text, text})
+      end)
+    end)
+  end
+
   defp do_stream(topic, item) do
     Logger.debug("Trying to push to #{topic}")
     Logger.debug("Pushing item to #{topic}")

lib/pleroma/web/views/streamer_view.ex

@@ -7,6 +7,7 @@ defmodule Pleroma.Web.StreamerView do
 
   alias Pleroma.Activity
   alias Pleroma.Conversation.Participation
+  alias Pleroma.Marker
   alias Pleroma.Notification
   alias Pleroma.User
   alias Pleroma.Web.MastodonAPI.NotificationView
@@ -164,6 +165,19 @@ defmodule Pleroma.Web.StreamerView do
     |> Jason.encode!()
   end
 
+  def render("marker.json", %Marker{} = marker) do
+    %{
+      event: "marker",
+      payload:
+        Pleroma.Web.MastodonAPI.MarkerView.render(
+          "markers.json",
+          markers: [marker]
+        )
+        |> Jason.encode!()
+    }
+    |> Jason.encode!()
+  end
+
   def render("pleroma_respond.json", %{type: type, result: result} = params) do
     %{
       event: "pleroma:respond",