Push SSE history updates to non-owner viewers

Owner dispatch only delivers history_update events to the user who owns
the changed history. Watching a shared or pinned non-owned history
(e.g. via the multi-history view) silently received nothing under SSE
because the dispatch path keys on history ownership and there was no
way for a viewer to opt in.

Add an additive viewer-subscription channel: the client POSTs the
history ids it wants pushes for to /api/events/history-subscriptions,
the dispatcher fans the record out to every webapp worker over Kombu,
and history_update fanout pushes to subscribed viewers in addition to
the owner. Subscriptions are refcounted on the client and replayed on
every EventSource onopen so reconnects don't drop them. Per-worker
subscription state is purged when a user's last connection on that
worker closes; the client's onopen replay re-asserts the desired set.

No authorization check on the dispatch path: SSE events only carry
history ids, the client follows up with a normal access-controlled
REST fetch, and leaking "history changed at T" pings is acceptable.

MultipleViewItem now subscribes whenever it renders a history the
current user does not own and SSE is enabled; owned histories continue
through owner dispatch with zero subscribe round-trip.

Tests: integration test exercising subscribe -> mutate -> receive ->
unsubscribe -> stop receiving end-to-end, plus vitest for the client
refcount and replay logic.

Author: mvdbeek

client/src/api/schema/schema.ts

@@ -1292,6 +1292,31 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/api/events/history-subscriptions": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        get?: never;
+        put?: never;
+        /**
+         * Subscribe to history_update SSE events for histories you don't own.
+         * @description Asks every webapp worker to start routing ``history_update`` events
+         *     for these histories to the requesting user/session, in addition to the
+         *     default owner-routing. Idempotent: re-subscribing to the same id is a
+         *     no-op. Clients re-send the full set after each ``EventSource.onopen``
+         *     so reconnects don't drop subscriptions.
+         */
+        post: operations["subscribe_history_viewer_api_events_history_subscriptions_post"];
+        /** Cancel viewer subscriptions for these histories. */
+        delete: operations["unsubscribe_history_viewer_api_events_history_subscriptions_delete"];
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/api/events/stream": {
         parameters: {
             query?: never;
@@ -15433,6 +15458,14 @@ export interface components {
              */
             url: string;
         };
+        /**
+         * HistoryViewerSubscriptionPayload
+         * @description REST payload for ``/api/events/history-subscriptions`` endpoints.
+         */
+        HistoryViewerSubscriptionPayload: {
+            /** History Ids */
+            history_ids: string[];
+        };
         /**
          * Hyperlink
          * @description Represents some text with an Hyperlink.
@@ -33478,6 +33511,92 @@ export interface operations {
             };
         };
     };
+    subscribe_history_viewer_api_events_history_subscriptions_post: {
+        parameters: {
+            query?: never;
+            header?: {
+                /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+                "run-as"?: string | null;
+            };
+            path?: never;
+            cookie?: never;
+        };
+        requestBody: {
+            content: {
+                "application/json": components["schemas"]["HistoryViewerSubscriptionPayload"];
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            204: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+            /** @description Request Error */
+            "4XX": {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["MessageExceptionModel"];
+                };
+            };
+            /** @description Server Error */
+            "5XX": {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["MessageExceptionModel"];
+                };
+            };
+        };
+    };
+    unsubscribe_history_viewer_api_events_history_subscriptions_delete: {
+        parameters: {
+            query?: never;
+            header?: {
+                /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+                "run-as"?: string | null;
+            };
+            path?: never;
+            cookie?: never;
+        };
+        requestBody: {
+            content: {
+                "application/json": components["schemas"]["HistoryViewerSubscriptionPayload"];
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            204: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+            /** @description Request Error */
+            "4XX": {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["MessageExceptionModel"];
+                };
+            };
+            /** @description Server Error */
+            "5XX": {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["MessageExceptionModel"];
+                };
+            };
+        };
+    };
     stream_events_api_events_stream_get: {
         parameters: {
             query?: never;

client/src/components/History/Multiple/MultipleViewItem.vue

@@ -3,10 +3,14 @@ import { faTimes } from "@fortawesome/free-solid-svg-icons";
 import { FontAwesomeIcon } from "@fortawesome/vue-fontawesome";
 import { BButton } from "bootstrap-vue";
 import { storeToRefs } from "pinia";
-import { computed, ref } from "vue";
+import { computed, ref, watch } from "vue";
 
+import { userOwnsHistory } from "@/api";
+import { useConfig } from "@/composables/config";
 import { useExtendedHistory } from "@/composables/detailedHistory";
+import { addHistoryViewerSubscription, removeHistoryViewerSubscription } from "@/composables/useNotificationSSE";
 import { useHistoryStore } from "@/stores/historyStore";
+import { useUserStore } from "@/stores/userStore";
 
 import CollectionPanel from "@/components/History/CurrentCollection/CollectionPanel.vue";
 import HistoryNavigation from "@/components/History/CurrentHistory/HistoryNavigation.vue";
@@ -24,6 +28,8 @@ const props = defineProps<Props>();
 
 const historyStore = useHistoryStore();
 const { currentHistoryId, pinnedHistories } = storeToRefs(historyStore);
+const { currentUser } = storeToRefs(useUserStore());
+const { config } = useConfig();
 
 const { history } = useExtendedHistory(props.source.id);
 
@@ -33,6 +39,31 @@ const sameToCurrent = computed(() => {
     return currentHistoryId.value === props.source.id;
 });
 
+// Owner routing covers histories the current user owns; only non-owned
+// histories need a viewer subscription. Skip in polling mode — there's no
+// SSE channel to push events to and the REST call would still hit the queue.
+const needsViewerSubscription = computed(() => {
+    if (!config.value?.enable_sse_updates) {
+        return false;
+    }
+    if (!history.value || !currentUser.value) {
+        return false;
+    }
+    return !userOwnsHistory(currentUser.value, history.value);
+});
+
+watch(
+    [needsViewerSubscription, () => props.source.id],
+    ([subscribe, id], _previous, onCleanup) => {
+        if (!subscribe || !id) {
+            return;
+        }
+        addHistoryViewerSubscription(id);
+        onCleanup(() => removeHistoryViewerSubscription(id));
+    },
+    { immediate: true },
+);
+
 function onViewCollection(collection: object) {
     selectedCollections.value = [...selectedCollections.value, collection];
 }

client/src/composables/useNotificationSSE.test.ts

@@ -0,0 +1,100 @@
+/**
+ * Unit tests for the viewer-subscription helpers in useNotificationSSE.
+ *
+ * The shared EventSource isn't exercised here — these tests focus on the
+ * client-side bookkeeping: refcounting, dedup, and HTTP shape. Reconnect
+ * replay is covered by an MSW request log assertion plus an explicit call
+ * into the (test-only) onopen replay path.
+ */
+
+import flushPromises from "flush-promises";
+import { http, HttpResponse } from "msw";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+
+import { useServerMock } from "@/api/client/__mocks__";
+
+import {
+    _resetHistoryViewerSubscriptionsForTest,
+    addHistoryViewerSubscription,
+    removeHistoryViewerSubscription,
+} from "./useNotificationSSE";
+
+interface SubscriptionRequest {
+    method: "POST" | "DELETE";
+    history_ids: string[];
+}
+
+const { server } = useServerMock();
+
+describe("useNotificationSSE viewer subscriptions", () => {
+    let requests: SubscriptionRequest[];
+
+    beforeEach(() => {
+        _resetHistoryViewerSubscriptionsForTest();
+        requests = [];
+        const recordHandler = async ({ request }: { request: Request }) => {
+            const body = (await request.json()) as { history_ids: string[] };
+            requests.push({
+                method: request.method as "POST" | "DELETE",
+                history_ids: body.history_ids,
+            });
+            return new HttpResponse(null, { status: 204 });
+        };
+        server.use(
+            http.post("/api/events/history-subscriptions", recordHandler),
+            http.delete("/api/events/history-subscriptions", recordHandler),
+        );
+    });
+
+    afterEach(() => {
+        _resetHistoryViewerSubscriptionsForTest();
+    });
+
+    it("POSTs once per first subscriber for a given history id", async () => {
+        addHistoryViewerSubscription("hist-A");
+        await flushPromises();
+        expect(requests).toHaveLength(1);
+        expect(requests[0]?.method).toBe("POST");
+        expect(requests[0]?.history_ids).toEqual(["hist-A"]);
+    });
+
+    it("refcounts duplicate subscriptions — second add is a no-op on the wire", async () => {
+        addHistoryViewerSubscription("hist-A");
+        addHistoryViewerSubscription("hist-A");
+        await flushPromises();
+        expect(requests.filter((r) => r.method === "POST")).toHaveLength(1);
+    });
+
+    it("only DELETEs when the last subscriber for an id releases", async () => {
+        addHistoryViewerSubscription("hist-A");
+        addHistoryViewerSubscription("hist-A");
+        await flushPromises();
+        const postCount = requests.filter((r) => r.method === "POST").length;
+
+        removeHistoryViewerSubscription("hist-A");
+        await flushPromises();
+        // First remove still has one outstanding refcount — must not DELETE yet.
+        expect(requests.filter((r) => r.method === "DELETE")).toHaveLength(0);
+        expect(requests.filter((r) => r.method === "POST")).toHaveLength(postCount);
+
+        removeHistoryViewerSubscription("hist-A");
+        await flushPromises();
+        const deletes = requests.filter((r) => r.method === "DELETE");
+        expect(deletes).toHaveLength(1);
+        expect(deletes[0]?.history_ids).toEqual(["hist-A"]);
+    });
+
+    it("ignores unsubscribes for ids that were never subscribed", async () => {
+        removeHistoryViewerSubscription("hist-never");
+        await flushPromises();
+        expect(requests).toHaveLength(0);
+    });
+
+    it("tracks distinct history ids independently", async () => {
+        addHistoryViewerSubscription("hist-A");
+        addHistoryViewerSubscription("hist-B");
+        await flushPromises();
+        const ids = requests.filter((r) => r.method === "POST").map((r) => r.history_ids[0]);
+        expect(new Set(ids)).toEqual(new Set(["hist-A", "hist-B"]));
+    });
+});

client/src/composables/useNotificationSSE.ts

@@ -78,6 +78,11 @@ function openSourceIfNeeded() {
         // Global readiness flag so Selenium tests can distinguish a working
         // SSE pipeline from the polling fallback.
         sseGlobals().__galaxy_sse_connected = true;
+        // Re-assert any viewer subscriptions the user accumulated. The server
+        // doesn't carry app-level subscription state across reconnects (it
+        // only knows the user from the cookie), so the client owns the source
+        // of truth and replays it on every successful open.
+        replayViewerSubscriptionsOnOpen();
     };
 
     sharedSource.onerror = () => {
@@ -207,3 +212,82 @@ export function useSSEConnectionStatus() {
         hasEverConnected: readonly(sseEverConnected),
     };
 }
+
+// ---------------------------------------------------------------------------
+// Viewer subscriptions for non-owned histories
+//
+// Owner routing already covers history_update events for histories the
+// current user owns. Watching a non-owned history (e.g. a shared history
+// pinned in the multi-history view) requires the client to opt in by POSTing
+// the history id to ``/api/events/history-subscriptions``. The server keeps a
+// per-worker map and pushes the same history_update events to every viewer
+// in that map. The desired set is held module-level so reconnects can
+// replay it and so multiple components subscribed to the same id only emit
+// one HTTP call.
+// ---------------------------------------------------------------------------
+
+const viewerSubscriptions = new Map<string, number>();
+
+function postViewerSubscription(method: "POST" | "DELETE", historyIds: string[]): Promise<void> {
+    if (historyIds.length === 0) {
+        return Promise.resolve();
+    }
+    return fetch(withPrefix("/api/events/history-subscriptions"), {
+        method,
+        headers: { "Content-Type": "application/json", Accept: "application/json" },
+        credentials: "same-origin",
+        body: JSON.stringify({ history_ids: historyIds }),
+    }).then((response) => {
+        if (!response.ok) {
+            throw new Error(`history-subscriptions ${method} failed: ${response.status}`);
+        }
+    });
+}
+
+function replayViewerSubscriptionsOnOpen(): void {
+    const ids = [...viewerSubscriptions.keys()];
+    if (ids.length === 0) {
+        return;
+    }
+    postViewerSubscription("POST", ids).catch((err) =>
+        console.error("Failed to replay history viewer subscriptions on reconnect:", err),
+    );
+}
+
+/**
+ * Add a viewer subscription for a history this user does not own. Refcounted
+ * so two components watching the same history share one server-side
+ * subscription and the first to mount opens it, last to unmount closes it.
+ *
+ * Idempotent per call: re-subscribing an already-tracked id does not POST a
+ * duplicate.
+ */
+export function addHistoryViewerSubscription(historyId: string): void {
+    const previous = viewerSubscriptions.get(historyId) ?? 0;
+    viewerSubscriptions.set(historyId, previous + 1);
+    if (previous === 0) {
+        postViewerSubscription("POST", [historyId]).catch((err) =>
+            console.error(`Failed to subscribe to history ${historyId}:`, err),
+        );
+    }
+}
+
+export function removeHistoryViewerSubscription(historyId: string): void {
+    const previous = viewerSubscriptions.get(historyId);
+    if (!previous) {
+        return;
+    }
+    if (previous > 1) {
+        viewerSubscriptions.set(historyId, previous - 1);
+        return;
+    }
+    viewerSubscriptions.delete(historyId);
+    postViewerSubscription("DELETE", [historyId]).catch((err) =>
+        console.error(`Failed to unsubscribe from history ${historyId}:`, err),
+    );
+}
+
+/** Test-only: drain the desired set so per-test state doesn't leak. */
+export function _resetHistoryViewerSubscriptionsForTest(): void {
+    viewerSubscriptions.clear();
+}