Renew the Mercure authorization instead of reconnecting without it

assets/js/src/core/modules/app/app-loader/app-loader.tsx

@@ -19,7 +19,7 @@ import { ErrorModalService } from '@Pimcore/modules/app/error-handler/services/e
 import { useIsAuthenticated } from '@Pimcore/modules/auth/hooks/use-is-authenticated'
 import { useTranslationLoader } from './loader/translation/loader'
 import { useUserLoader } from './loader/user/loader'
-import { useMercureCreateCookieMutation } from '../mercure-api-slice.gen'
+import { renewMercureAuthorization } from '@Pimcore/modules/app/mercure/mercure-authorization'
 import { useSettingsLoader } from './loader/settings/loader'
 import { useActiveBundlesLoader } from './loader/active-bundles/loader'
 import { useLanguageLoader } from './loader/language/loader'
@@ -91,7 +91,6 @@ export const AppLoader = (props: IAppLoaderProps): React.JSX.Element => {
 
   const { loadPublicTranslations, loadTranslations } = useTranslationLoader()
   const { loadUser } = useUserLoader()
-  const [fetchMercureCookie] = useMercureCreateCookieMutation()
   const { loadSettings } = useSettingsLoader()
   const { loadActiveBundles } = useActiveBundlesLoader()
   const { loadAvailableLocales } = useLanguageLoader()
@@ -137,8 +136,12 @@ export const AppLoader = (props: IAppLoaderProps): React.JSX.Element => {
         // silently drops every PRIVATE update on the per-user topic
         // (studio-backend-default/user/{id}) for the whole session — until the
         // next reload, where the cookie already exists. Fetch the cookie first,
-        // then start the subscription.
-        await fetchMercureCookie()
+        // then start the subscription. This also teaches the renewal how long the
+        // cookie lives; every later reconnect renews it on its own.
+        await renewMercureAuthorization().catch(() => {
+          // Studio still has to load. Live updates stay off until a reconnect
+          // manages to authorize again, which every reconnect attempts.
+        })
 
         const user = selectCurrentUser(store.getState())
         if (!isNil(user?.id)) {

assets/js/src/core/modules/app/app-loader/loader/global-message-bus/loader.ts

@@ -14,6 +14,7 @@ import { type GlobalMessageBus } from '@Pimcore/modules/global-message-bus/servi
 import { type GlobalMessageBusProcess } from '@Pimcore/modules/background-processor/process/global-message-bus-process'
 import { type BackgroundProcessor } from '@Pimcore/modules/background-processor/services/background-processor'
 import { topics } from '@Pimcore/modules/execution-engine/topics'
+import { startMercureAuthorizationRenewal } from '@Pimcore/modules/app/mercure/mercure-authorization'
 
 interface UseGlobalMessageBusLoaderReturn {
   initGlobalMessageBus: (userId: number) => void
@@ -39,6 +40,10 @@ export const useGlobalMessageBusLoader = (): UseGlobalMessageBusLoaderReturn =>
       backgroundProcessor.registerProcess(globalProcess)
       messageRegistry.startGlobalSubscription()
 
+      // Backstop for hubs that never close a stream: the subscription is re-authorised on every
+      // reconnect, but without a write timeout a single connection can outlive the cookie.
+      startMercureAuthorizationRenewal()
+
       // Reconnect when tab becomes visible again
       document.addEventListener('visibilitychange', () => {
         if (document.visibilityState === 'visible' && !globalProcess.isConnected()) {

assets/js/src/core/modules/app/mercure/mercure-authorization.test.ts

@@ -0,0 +1,207 @@
+/**
+ * This source file is available under the terms of the
+ * Pimcore Open Core License (POCL)
+ * Full copyright and license information is available in
+ * LICENSE.md which is distributed with this source code.
+ *
+ *  @copyright  Copyright (c) Pimcore GmbH (https://www.pimcore.com)
+ *  @license    Pimcore Open Core License (POCL)
+ */
+
+import type * as MercureAuthorizationModule from './mercure-authorization'
+
+const dispatch = jest.fn()
+const initiate = jest.fn(() => 'mercure-create-cookie-action')
+
+jest.mock('@Pimcore/app/store', () => ({
+  store: { dispatch: (action: unknown) => dispatch(action) }
+}))
+
+jest.mock('@Pimcore/modules/app/mercure-api-slice.gen', () => ({
+  api: { endpoints: { mercureCreateCookie: { initiate: () => initiate() } } }
+}))
+
+type AuthorizationModule = typeof MercureAuthorizationModule
+
+const respondWith = (response: unknown): void => {
+  dispatch.mockReturnValue({ unwrap: async () => response })
+}
+
+const rejectWith = (error: Error): void => {
+  dispatch.mockReturnValue({ unwrap: async () => { throw error } })
+}
+
+const loadModule = async (): Promise<AuthorizationModule> => {
+  let module: AuthorizationModule | undefined
+  await jest.isolateModulesAsync(async () => {
+    module = await import('./mercure-authorization')
+  })
+
+  return module!
+}
+
+describe('mercure authorization', () => {
+  beforeEach(() => {
+    jest.useFakeTimers()
+    dispatch.mockReset()
+    initiate.mockClear()
+    respondWith({ cookieLifetime: 3600 })
+  })
+
+  afterEach(() => {
+    jest.useRealTimers()
+  })
+
+  it('requests a new cookie and learns its lifetime from the response', async () => {
+    const module = await loadModule()
+
+    await module.renewMercureAuthorization()
+
+    expect(initiate).toHaveBeenCalledTimes(1)
+    expect(module.getAuthorizationLifetimeMs()).toBe(3_600_000)
+  })
+
+  it('keeps the last known lifetime when the response does not carry one', async () => {
+    const module = await loadModule()
+    respondWith({ cookieLifetime: 900 })
+    await module.renewMercureAuthorization()
+
+    respondWith(undefined)
+    await module.renewMercureAuthorization()
+
+    expect(module.getAuthorizationLifetimeMs()).toBe(900_000)
+  })
+
+  it('rejects when the request fails, so the caller can fall back to the cookie it has', async () => {
+    const module = await loadModule()
+    rejectWith(new Error('offline'))
+
+    await expect(module.renewMercureAuthorization()).rejects.toThrow('offline')
+  })
+
+  it('renews in the background before the cookie expires, and keeps renewing', async () => {
+    const module = await loadModule()
+    respondWith({ cookieLifetime: 1000 })
+    // The app loader fetches the first cookie through the same call, which is what teaches the
+    // schedule the configured lifetime instead of the fallback.
+    await module.renewMercureAuthorization()
+    initiate.mockClear()
+
+    module.startMercureAuthorizationRenewal()
+
+    // Nothing yet: renewal happens at 80% of the lifetime, not immediately.
+    await jest.advanceTimersByTimeAsync(799_000)
+    expect(initiate).not.toHaveBeenCalled()
+
+    await jest.advanceTimersByTimeAsync(1_000)
+    expect(initiate).toHaveBeenCalledTimes(1)
+
+    await jest.advanceTimersByTimeAsync(800_000)
+    expect(initiate).toHaveBeenCalledTimes(2)
+  })
+
+  it('keeps the schedule alive when a renewal fails', async () => {
+    const module = await loadModule()
+    rejectWith(new Error('offline'))
+
+    module.startMercureAuthorizationRenewal()
+
+    await jest.advanceTimersByTimeAsync(2_880_000)
+    expect(initiate).toHaveBeenCalledTimes(1)
+
+    await jest.advanceTimersByTimeAsync(2_880_000)
+    expect(initiate).toHaveBeenCalledTimes(2)
+  })
+
+  it('does not stack a second timer when started twice', async () => {
+    const module = await loadModule()
+
+    module.startMercureAuthorizationRenewal()
+    module.startMercureAuthorizationRenewal()
+
+    await jest.advanceTimersByTimeAsync(2_880_000)
+
+    expect(initiate).toHaveBeenCalledTimes(1)
+  })
+
+  it('stops renewing once cancelled', async () => {
+    const module = await loadModule()
+
+    module.startMercureAuthorizationRenewal()
+    module.stopMercureAuthorizationRenewal()
+
+    await jest.advanceTimersByTimeAsync(2_880_000)
+
+    expect(initiate).not.toHaveBeenCalled()
+  })
+
+  it('stops renewing when cancelled while a renewal is in flight', async () => {
+    const module = await loadModule()
+    let release = (): void => {}
+    dispatch.mockReturnValue({
+      unwrap: async () => await new Promise(resolve => { release = () => { resolve({}) } })
+    })
+
+    module.startMercureAuthorizationRenewal()
+    await jest.advanceTimersByTimeAsync(2_880_000)
+    expect(initiate).toHaveBeenCalledTimes(1)
+
+    // Cancelled while the request is still open: the schedule must not re-arm when it settles.
+    module.stopMercureAuthorizationRenewal()
+    release()
+    await jest.advanceTimersByTimeAsync(2_880_000 * 3)
+
+    expect(initiate).toHaveBeenCalledTimes(1)
+  })
+
+  it('shares one request between concurrent renewals', async () => {
+    const module = await loadModule()
+    let release = (): void => {}
+    dispatch.mockReturnValue({
+      unwrap: async () => await new Promise(resolve => { release = () => { resolve({ cookieLifetime: 3600 }) } })
+    })
+
+    const first = module.renewMercureAuthorization()
+    const second = module.renewMercureAuthorization()
+    release()
+    await Promise.all([first, second])
+
+    expect(initiate).toHaveBeenCalledTimes(1)
+  })
+
+  it('reports the cookie as invalid until the first renewal succeeded', async () => {
+    const module = await loadModule()
+
+    expect(module.isAuthorizationValid()).toBe(false)
+    expect(module.isAuthorizationFresh()).toBe(false)
+
+    await module.renewMercureAuthorization()
+
+    expect(module.isAuthorizationValid()).toBe(true)
+    expect(module.isAuthorizationFresh()).toBe(true)
+  })
+
+  it('reports the cookie as stale but still valid inside the last fifth of its lifetime', async () => {
+    const module = await loadModule()
+    respondWith({ cookieLifetime: 1000 })
+    await module.renewMercureAuthorization()
+
+    await jest.advanceTimersByTimeAsync(801_000)
+    expect(module.isAuthorizationFresh()).toBe(false)
+    expect(module.isAuthorizationValid()).toBe(true)
+
+    await jest.advanceTimersByTimeAsync(200_000)
+    expect(module.isAuthorizationValid()).toBe(false)
+  })
+
+  it('keeps the previous expiry when a renewal fails', async () => {
+    const module = await loadModule()
+    respondWith({ cookieLifetime: 1000 })
+    await module.renewMercureAuthorization()
+
+    rejectWith(new Error('offline'))
+    await expect(module.renewMercureAuthorization()).rejects.toThrow('offline')
+
+    expect(module.isAuthorizationValid()).toBe(true)
+  })
+})