From 6f3cb4a3526864a99f9a3c2c23388f938504cdd3 Mon Sep 17 00:00:00 2001 From: Maxime Date: Thu, 3 Sep 2026 15:28:46 +0200 Subject: [PATCH 1/4] feat: migrate to the Mercure 1.0 protocol MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Test server now runs mercure/caddy v1.0.0-alpha.3. The 2.8 builder ships Go 1.23.4 and cannot build it (requires go >= 1.26), so the builder moves to 2.11.4 — the Caddy version mercure/caddy pins — and both plugins are pinned explicitly: unpinned, xcaddy resolved mercure/caddy to the latest stable tag and silently gave 0.24.2. Caddyfile ported to modern mode, directive for directive: flag-form anonymous, `debugger` for the renamed `ui`, and an issuer block replacing the flat publisher_jwt/subscriber_jwt, which now work only under protocol_version_compatibility. resource_identifier is set because an identifier ending in /.well-known/mercure also becomes the base URL relative topics resolve against, and the topics here are rel="self" link values. Client subscribes with `match` instead of the removed `topic` parameter. The tracked topics are those same relative link values, so the exact matcher is the right semantics; match_urlpattern is not exposed yet. Publisher fixtures reminted as RFC 9068 access tokens: typ at+jwt, iss, aud, exp and authorization_details. The legacy `mercure` claim and typ JWT are both rejected in modern mode. --- Caddyfile | 26 ++++++++++++++++++++++---- Dockerfile | 8 +++++--- packages/mercure/mercure.ts | 2 +- tests-server/github.html | 2 +- tests-server/mercure.html | 2 +- tests/mercure.spec.ts | 6 +++--- 6 files changed, 33 insertions(+), 13 deletions(-) diff --git a/Caddyfile b/Caddyfile index aeb7a79..7d16573 100644 --- a/Caddyfile +++ b/Caddyfile @@ -43,8 +43,26 @@ handle / { } mercure { - publisher_jwt key - subscriber_jwt key - anonymous true - ui true + anonymous + + # Mercure 1.0 modern mode: access tokens are RFC 9068 JWTs bound to an issuer + # and to this hub's resource identifier. An identifier ending in + # /.well-known/mercure doubles as the base URL relative topics resolve + # against, which is what our rel="self" link values are. + resource_identifier https://localhost/.well-known/mercure + issuer https://localhost { + publisher { + jwt key HS256 + } + subscriber { + jwt key HS256 + } + } + + # INSECURE: dev only. Serves the hub's debugger UI at + # /.well-known/mercure/debug/. This is the 1.0 name of the former "ui" + # directive. Not "playground": that one also forces cors_origins and + # publish_origins to "*" and drops the cookie name prefix, none of which this + # same-origin test server needs. + debugger } diff --git a/Dockerfile b/Dockerfile index 93a57db..e1f2fa7 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,7 +1,9 @@ -FROM caddy:2.8-builder AS builder +FROM caddy:2.11.4-builder AS builder -RUN xcaddy build --with github.com/dunglas/mercure/caddy --with github.com/dunglas/vulcain/caddy +RUN xcaddy build \ + --with github.com/dunglas/mercure/caddy@v1.0.0-alpha.3 \ + --with github.com/dunglas/vulcain/caddy@v1.4.3 -FROM caddy:2.8 AS app_server +FROM caddy:2.11.4 AS app_server COPY --from=builder /usr/bin/caddy /usr/bin/caddy diff --git a/packages/mercure/mercure.ts b/packages/mercure/mercure.ts index 1a5e941..5a2ddfc 100644 --- a/packages/mercure/mercure.ts +++ b/packages/mercure/mercure.ts @@ -26,7 +26,7 @@ function listen(mercureUrl: string, options: Options = {}) { const url = new URL(mercureUrl) topics.forEach((_, topic) => { - url.searchParams.append('topic', topic) + url.searchParams.append('match', topic) }) const headers: {[key: string]: string} = options.headers || {} diff --git a/tests-server/github.html b/tests-server/github.html index fdad921..637b48d 100644 --- a/tests-server/github.html +++ b/tests-server/github.html @@ -74,7 +74,7 @@ // You can pass formData as a fetch body directly: fetch(form.action, { method: form.method, body: body.toString(), headers: { 'Content-Type': 'application/x-www-form-urlencoded', - 'Authorization': 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJtZXJjdXJlIjp7InN1YnNjcmliZSI6WyIqIl0sInB1Ymxpc2giOlsiKiJdfX0.NXhzhXJ8VTxiRRW3pAB4EgP7s_guZeibwzAGw3wZ_KY' + 'Authorization': 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6ImF0K2p3dCJ9.eyJpc3MiOiJodHRwczovL2xvY2FsaG9zdCIsImF1ZCI6Imh0dHBzOi8vbG9jYWxob3N0Ly53ZWxsLWtub3duL21lcmN1cmUiLCJzdWIiOiJlc2EtdGVzdC1zZXJ2ZXIiLCJjbGllbnRfaWQiOiJlc2EtdGVzdC1zZXJ2ZXIiLCJpYXQiOjE3ODg0MjU1MDUsImV4cCI6NDEwMjQ0NDgwMCwiYXV0aG9yaXphdGlvbl9kZXRhaWxzIjpbeyJ0eXBlIjoiaHR0cHM6Ly9tZXJjdXJlLnJvY2tzL2F1dGhvcml6YXRpb24tZGV0YWlsIiwiYWN0aW9ucyI6WyJwdWJsaXNoIl0sInRvcGljcyI6W3sibWF0Y2giOiIqIn1dfV19.825udJ6bE3p1HpZM_1QZ83DVpOTtuvGE_cQvJby7d0k' } }); } diff --git a/tests-server/mercure.html b/tests-server/mercure.html index d162ad8..5325d7e 100644 --- a/tests-server/mercure.html +++ b/tests-server/mercure.html @@ -53,7 +53,7 @@ // You can pass formData as a fetch body directly: fetch(form.action, { method: form.method, body: body.toString(), headers: { 'Content-Type': 'application/x-www-form-urlencoded', - 'Authorization': 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJtZXJjdXJlIjp7InN1YnNjcmliZSI6WyIqIl0sInB1Ymxpc2giOlsiKiJdfX0.NXhzhXJ8VTxiRRW3pAB4EgP7s_guZeibwzAGw3wZ_KY' + 'Authorization': 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6ImF0K2p3dCJ9.eyJpc3MiOiJodHRwczovL2xvY2FsaG9zdCIsImF1ZCI6Imh0dHBzOi8vbG9jYWxob3N0Ly53ZWxsLWtub3duL21lcmN1cmUiLCJzdWIiOiJlc2EtdGVzdC1zZXJ2ZXIiLCJjbGllbnRfaWQiOiJlc2EtdGVzdC1zZXJ2ZXIiLCJpYXQiOjE3ODg0MjU1MDUsImV4cCI6NDEwMjQ0NDgwMCwiYXV0aG9yaXphdGlvbl9kZXRhaWxzIjpbeyJ0eXBlIjoiaHR0cHM6Ly9tZXJjdXJlLnJvY2tzL2F1dGhvcml6YXRpb24tZGV0YWlsIiwiYWN0aW9ucyI6WyJwdWJsaXNoIl0sInRvcGljcyI6W3sibWF0Y2giOiIqIn1dfV19.825udJ6bE3p1HpZM_1QZ83DVpOTtuvGE_cQvJby7d0k' } }); } diff --git a/tests/mercure.spec.ts b/tests/mercure.spec.ts index 84c6829..c4e8d90 100644 --- a/tests/mercure.spec.ts +++ b/tests/mercure.spec.ts @@ -10,15 +10,15 @@ test('mercure', async ({ page }) => { num++ } - if (request.url().startsWith('https://localhost/.well-known/mercure?topic=%2Fauthors%2F1')) { + if (request.url().startsWith('https://localhost/.well-known/mercure?match=%2Fauthors%2F1')) { requestedMercure = true } - if (request.url().startsWith('https://localhost/.well-known/mercure?topic=%2Fauthors%2F1&topic=%2Fauthors%2F2')) { + if (request.url().startsWith('https://localhost/.well-known/mercure?match=%2Fauthors%2F1&match=%2Fauthors%2F2')) { subscribedToBoth = true } - if (request.url().startsWith('https://localhost/.well-known/mercure?topic=%2Fauthors%2F2')) { + if (request.url().startsWith('https://localhost/.well-known/mercure?match=%2Fauthors%2F2')) { unsubscribedAuthor1 = true } }) From aa5e68f676fc32ee9e9de09edddb276c4ddf2d7b Mon Sep 17 00:00:00 2001 From: Maxime Date: Tue, 8 Sep 2026 16:44:57 +0200 Subject: [PATCH 2/4] fix(mercure): send the resume cursor as a query parameter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every listen() opens a fresh connection, so the id of the last update has to travel with the request. The client only set the Last-Event-Id header, which a native EventSource cannot do — resuming worked solely because of the npm eventsource polyfill, even though the README invites callers to pass their own EventSource. The spec is explicit that the hub "MUST then take the union" of the query and body components and that this "applies to the topic matcher parameters and to last_event_id ... alike", so the cursor now goes in the query too. The header is still sent, for implementations that support it. --- packages/mercure/README.md | 4 ++++ packages/mercure/mercure.ts | 9 +++++++++ 2 files changed, 13 insertions(+) diff --git a/packages/mercure/README.md b/packages/mercure/README.md index a1e67e7..99b3c31 100644 --- a/packages/mercure/README.md +++ b/packages/mercure/README.md @@ -54,6 +54,10 @@ Available options: This can be used in conjunction with [@api-platform/ld](/linked-data) as the `fetchFn`. +### Resuming after a disconnection + +The id of the last update received is kept and sent back when a subscription is rebuilt, as both the `last_event_id` query parameter and the `Last-Event-Id` request header. The query parameter is what makes this work with a native `EventSource`, which cannot set headers. + ### Examples See [our Tanstack query example](https://github.com/api-platform/esa/blob/main/tests-server/mercure.html) or the source code of our [home page](https://github.com/api-platform/esa/blob/main/api/public/index.js). diff --git a/packages/mercure/mercure.ts b/packages/mercure/mercure.ts index 5a2ddfc..8d3428a 100644 --- a/packages/mercure/mercure.ts +++ b/packages/mercure/mercure.ts @@ -31,6 +31,15 @@ function listen(mercureUrl: string, options: Options = {}) { const headers: {[key: string]: string} = options.headers || {} if (lastEventId) { + // Every call here opens a fresh connection, so the cursor has to travel + // with the request. A native EventSource cannot set headers, hence the + // query parameter: the hub takes the union of the query and body + // components, and last_event_id is single-valued. The header is sent too, + // for EventSource implementations that support it and for the automatic + // reconnections they perform on their own. + url.searchParams.append('last_event_id', lastEventId) + // The request header keeps its name in 1.0; only the hub's response header + // was renamed to Mercure-Last-Event-ID. headers['Last-Event-Id'] = lastEventId } From 88df95d4d5367805cdeca04046afd10d3c519444 Mon Sep 17 00:00:00 2001 From: Maxime Date: Thu, 3 Sep 2026 15:39:06 +0200 Subject: [PATCH 3/4] feat(mercure): subscribe with a URL Pattern MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Collapsing a family of topics into one subscription is the headline feature of Mercure 1.0, and the client had no way to reach it. A new `matchUrlPattern` option subscribes with `match_urlpattern` instead of the exact `rel="self"` topic: const matchUrlPattern = '/authors/:id' await mercure('/authors/1', {matchUrlPattern, onUpdate}) await mercure('/authors/2', {matchUrlPattern, onUpdate}) // reuses it Three changes were needed beyond sending a different query parameter: Subscriptions are now keyed by matcher — an exact topic, or a pattern — with a second map from topic to matcher, because a pattern stands for many topics. close() is reference counted. close('/authors/1') must not tear down a subscription /authors/2 is still using, so a subscription lives until the last topic it covers is closed. An exact matcher holds exactly one topic, so its behaviour is unchanged. Joining an already-subscribed matcher no longer reconnects. Closing and reopening the EventSource for a subscription URL that did not change defeats the point of collapsing the family, so the callbacks are rebound on the live connection and the most recent registration serves the stream. listen() also used to append every known topic to the subscription URL regardless of which hub it belonged to; it now filters by hub. Invisible with one hub, wrong with two. The new page and spec assert what the feature promises: two resources of one family produce a single `match_urlpattern` subscription and no `match` parameter, an update on a topic that was never fetched still arrives, and the connection count stays at one. Both are plain module JS — no JSX, no esm.sh/run, so they cannot rot the way the React pages did. README rewritten for 1.0: it documented the removed `topic` parameter, and its example awaited a response and then called .then() on it. --- packages/mercure/README.md | 42 ++++++- packages/mercure/mercure.ts | 160 +++++++++++++++++++++------ tests-server/mercure-urlpattern.html | 71 ++++++++++++ tests/mercure-urlpattern.spec.ts | 33 ++++++ 4 files changed, 265 insertions(+), 41 deletions(-) create mode 100644 tests-server/mercure-urlpattern.html create mode 100644 tests/mercure-urlpattern.spec.ts diff --git a/packages/mercure/README.md b/packages/mercure/README.md index 99b3c31..7a75ac8 100644 --- a/packages/mercure/README.md +++ b/packages/mercure/README.md @@ -1,6 +1,8 @@ # @api-platform/mercure -`@api-platform/mercure` is an EventSource wrapper that [discovers a Mercure Hub](https://mercure.rocks/spec#discovery) according to the Link headers and handles subscriptions for you. +`@api-platform/mercure` is an EventSource wrapper that [discovers a Mercure Hub](https://mercure.rocks/docs/1.0/concepts/discovery) according to the Link headers and handles subscriptions for you. + +It speaks the [Mercure 1.0](https://mercure.rocks/docs/1.0/introduction) protocol: subscriptions are sent as `match` (exact) or `match_urlpattern` (URL Pattern) query parameters. The pre-1.0 `topic` parameter is not supported. ```javascript import mercure, { close } from "@api-platform/mercure"; @@ -9,7 +11,7 @@ const res = await mercure('https://localhost/authors/1', { onUpdate: (author) => console.log(author) }) -const author = res.then(res => res.json()) +const author = await res.json() // Close if you need to history.onpushstate = function(e) { @@ -24,7 +26,7 @@ Link: ; rel="self" Link: ; rel="mercure" ``` -A new `EventSource` is created by subscribing to the topic `https://localhost/authors/1` on the Hub `https://localhost/.well-known/mercure`. +A new `EventSource` is created by subscribing to the topic `https://localhost/authors/1` on the Hub `https://localhost/.well-known/mercure`, as `?match=https%3A%2F%2Flocalhost%2Fauthors%2F1`. ## Installation @@ -43,21 +45,51 @@ const res = await mercure('https://localhost/authors/1', { onUpdate: (author) => console.log(author) }) -const author = res.then(res => res.json()) +const author = await res.json() ``` Available options: +- `onUpdate` called with each update, parsed as JSON unless `rawEvent` is set +- `rawEvent` to receive the `MessageEvent` instead of the parsed payload - `onError` on EventSource error callback - `EventSource` to provide your own `EventSource` constructor +- `withCredentials` to send credentials with the subscription, `true` by default - `fetchFn` to provide your own fetch function, it needs to return a response so that we can read headers +- `matchUrlPattern` to subscribe with a URL Pattern instead of the exact topic, see below This can be used in conjunction with [@api-platform/ld](/linked-data) as the `fetchFn`. +### Subscribing to a family of topics + +By default each resource gets its own exact subscription. Fetching one hundred authors means one hundred `match` parameters on the subscription URL. + +`matchUrlPattern` collapses them into one. Pass the [URL Pattern](https://mercure.rocks/docs/1.0/concepts/topics-and-matchers) covering the family, and every resource it matches shares a single subscription: + +```javascript +import mercure, { close } from "@api-platform/mercure"; + +const matchUrlPattern = '/authors/:id' + +await mercure('/authors/1', {matchUrlPattern, onUpdate}) +// Reuses the subscription above. The hub sees one `match_urlpattern=/authors/:id`, +// not two `match=` parameters, and the connection is never dropped. +await mercure('/authors/2', {matchUrlPattern, onUpdate}) +``` + +URL Patterns support named groups (`:id`), wildcards (`*`), regular expression constraints (`:type(news|alerts)`) and optional segments (`/items{/:tail}?`). + +Two consequences worth knowing: + +- **You receive updates for topics you never fetched.** The pattern is what the hub matches against, so `/authors/3` reaches you even if you only ever fetched authors 1 and 2. That is the point, but it means the payload is the only thing that tells updates apart: an SSE frame carries `id`, `event` and `data`, never the topic. With JSON-LD, dispatch on `@id`. +- **`close(topic)` is reference counted.** The subscription stays open while any of the topics it covers is still in use, and is dropped once the last one closes. + +When several resources share a matcher, the callbacks passed to the most recent `mercure()` call serve the stream — the connection is reused, not rebuilt. + ### Resuming after a disconnection The id of the last update received is kept and sent back when a subscription is rebuilt, as both the `last_event_id` query parameter and the `Last-Event-Id` request header. The query parameter is what makes this work with a native `EventSource`, which cannot set headers. ### Examples -See [our Tanstack query example](https://github.com/api-platform/esa/blob/main/tests-server/mercure.html) or the source code of our [home page](https://github.com/api-platform/esa/blob/main/api/public/index.js). +See [our Tanstack query example](https://github.com/api-platform/esa/blob/main/tests-server/mercure.html), the [URL Pattern example](https://github.com/api-platform/esa/blob/main/tests-server/mercure-urlpattern.html), or the source code of our [home page](https://github.com/api-platform/esa/blob/main/api/public/index.js). diff --git a/packages/mercure/mercure.ts b/packages/mercure/mercure.ts index 8d3428a..d5df896 100644 --- a/packages/mercure/mercure.ts +++ b/packages/mercure/mercure.ts @@ -1,7 +1,14 @@ import {EventSource} from 'eventsource' -let lastEventId: string -const eventSources = new Map(); -const topics = new Map(); + +// Mercure 1.0 encodes the matcher type in the name of the query parameter: +// bare "match" selects the default "exact" type, "match_urlpattern" selects +// URL Patterns (WHATWG), which stand for a whole family of topics. +type MatcherType = 'exact' | 'urlpattern' + +const matcherParam: Record = { + exact: 'match', + urlpattern: 'match_urlpattern', +} type Options = { rawEvent?: boolean; @@ -11,24 +18,69 @@ type Options = { onError?: (error: unknown) => void; onUpdate?: (data: MessageEvent|T) => void; withCredentials?: boolean; + // Subscribe with a URL Pattern instead of the exact "rel=self" topic. Every + // resource whose topic this pattern covers then shares a single + // subscription: "/authors/:id" replaces one subscription per author. + matchUrlPattern?: string; } & RequestInit; -function listen(mercureUrl: string, options: Options = {}) { - if (eventSources.has(mercureUrl)) { - const eventSource = eventSources.get(mercureUrl) - eventSource.eventSource.close() - eventSources.delete(mercureUrl) +type Subscription = { + mercureUrl: string; + type: MatcherType; + // The topics this matcher currently stands for. An exact matcher holds one; + // a URL Pattern holds every fetched resource it covers, so the subscription + // outlives close() on any single one of them. + topics: Set; +} + +let lastEventId: string +const eventSources = new Map(); +// Matcher (an exact topic, or a URL Pattern) -> the subscription it opens. +const subscriptions = new Map(); +// Topic -> the matcher covering it. +const matchers = new Map(); + +// Attach the callbacks to a connection. Split out of listen() so a new +// subscriber joining an existing matcher can refresh them without dropping +// the stream and reconnecting. +function bind(entry: {eventSource: any, options: Options}, options: Options) { + entry.options = options + entry.eventSource.onmessage = (event: MessageEvent) => { + lastEventId = event.lastEventId + if (options.onUpdate) { + try { + options.onUpdate(options.rawEvent ? event : JSON.parse(event.data)) + } catch (e) { + options.onError && options.onError(e) + } + } } - if (topics.size === 0) { - return; + entry.eventSource.onerror = options.onError +} + +function listen(mercureUrl: string, options: Options = {}) { + const current = eventSources.get(mercureUrl) + if (current) { + current.eventSource.close() + eventSources.delete(mercureUrl) } const url = new URL(mercureUrl) - topics.forEach((_, topic) => { - url.searchParams.append('match', topic) + let subscribed = 0 + subscriptions.forEach((subscription, matcher) => { + if (subscription.mercureUrl !== mercureUrl) { + return + } + + url.searchParams.append(matcherParam[subscription.type], matcher) + subscribed++ }) + if (subscribed === 0) { + return; + } + const headers: {[key: string]: string} = options.headers || {} if (lastEventId) { // Every call here opens a fresh connection, so the cursor has to travel @@ -44,33 +96,33 @@ function listen(mercureUrl: string, options: Options = {}) { } const eventSource = new (options.EventSource ?? EventSource)(url.toString(), { withCredentials: options.withCredentials !== undefined ? options.withCredentials : true, headers}); - eventSource.onmessage = (event: MessageEvent) => { - lastEventId = event.lastEventId - if (options.onUpdate) { - try { - options.onUpdate(options.rawEvent ? event : JSON.parse(event.data)) - } catch (e) { - options.onError && options.onError(e) - } - } - } - - eventSource.onerror = options.onError - eventSources.set(mercureUrl, { - options: options, - eventSource: eventSource - }) + const entry = {options, eventSource} + bind(entry, options) + eventSources.set(mercureUrl, entry) } export function close(topic: string) { - if (!topics.has(topic)) { + const matcher = matchers.get(topic) + if (matcher === undefined) { + return + } + + matchers.delete(topic) + + const subscription = subscriptions.get(matcher) + if (!subscription) { return } - const mercureUrl = topics.get(topic) - topics.delete(topic) - const ee = eventSources.get(mercureUrl) - listen(mercureUrl, ee.options) + subscription.topics.delete(topic) + // A URL Pattern covers a family: keep the subscription as long as one of its + // topics is still in use. + if (subscription.topics.size > 0) { + return + } + + subscriptions.delete(matcher) + listen(subscription.mercureUrl, eventSources.get(subscription.mercureUrl)?.options) } export default async function mercure(url: string, opts: Options) { @@ -93,12 +145,48 @@ export default async function mercure(url: string, opts: Options) { } }); - if (mercureUrl) { - topics.set(topic === undefined ? url : topic, mercureUrl) + if (!mercureUrl) { + return res + } + + topic = topic === undefined ? url : topic + const matcher = opts.matchUrlPattern ?? topic + + // Moving a topic from one matcher to another: release the old one first, + // otherwise it keeps a topic nothing will ever close. + const previous = matchers.get(topic) + if (previous !== undefined && previous !== matcher) { + close(topic) + } + + let subscription = subscriptions.get(matcher) + const opened = subscription === undefined + + if (subscription === undefined) { + subscription = { + mercureUrl, + type: opts.matchUrlPattern === undefined ? 'exact' : 'urlpattern', + topics: new Set(), + } + subscriptions.set(matcher, subscription) + } + + subscription.topics.add(topic) + matchers.set(topic, matcher) + + const entry = eventSources.get(mercureUrl) + if (opened || !entry) { listen(mercureUrl, opts) + + return res } + // The matcher is already subscribed, so this resource needs no new + // subscription at all — that is the point of collapsing a family into + // one URL Pattern. Refresh the callbacks in place instead of + // reconnecting; the latest registration serves the stream. + bind(entry, opts) + return res; }); } - diff --git a/tests-server/mercure-urlpattern.html b/tests-server/mercure-urlpattern.html new file mode 100644 index 0000000..d59f076 --- /dev/null +++ b/tests-server/mercure-urlpattern.html @@ -0,0 +1,71 @@ + + + + + + Mercure URL Pattern + + + + +

+ + +
    + + + + + diff --git a/tests/mercure-urlpattern.spec.ts b/tests/mercure-urlpattern.spec.ts new file mode 100644 index 0000000..261a992 --- /dev/null +++ b/tests/mercure-urlpattern.spec.ts @@ -0,0 +1,33 @@ +import { test, expect } from '@playwright/test'; + +const hub = 'https://localhost/.well-known/mercure?' + +test('mercure url pattern', async ({ page }) => { + const subscriptions: string[] = [] + page.on('request', request => { + const url = request.url() + if (request.method() === 'GET' && url.startsWith(hub)) { + subscriptions.push(url.substring(hub.length)) + } + }) + + await page.goto('https://localhost/mercure-urlpattern'); + await expect(page.getByTestId('ready')).toHaveText('ready'); + + // Two resources of the same family, and a single subscription expressed as + // one URL Pattern — no `match=` subscription at all. + expect(subscriptions).toHaveLength(1); + expect(subscriptions[0]).toContain('match_urlpattern=%2Fauthors%2F%3Aid'); + expect(subscriptions[0]).not.toContain('match='); + + // An update on a topic the page did fetch. + await page.getByTestId('publish-1').click(); + await expect(page.getByTestId('received')).toContainText('/authors/1: Soyuka'); + + // And one on a topic it never fetched: the pattern covers the whole family. + await page.getByTestId('publish-3').click(); + await expect(page.getByTestId('received')).toContainText('/authors/3: Emily Rodda'); + + // Still one connection: joining an existing matcher must not reconnect. + expect(subscriptions).toHaveLength(1); +}); From b0bd7fe6072cdd9ec6608e6d829f4f8c5ad9d65a Mon Sep 17 00:00:00 2001 From: Maxime Date: Fri, 4 Sep 2026 10:47:55 +0200 Subject: [PATCH 4/4] refactor(mercure): key subscription state by hub Subscriptions, the resume cursor and the connection lived in module-wide maps. listen() appended every known topic to the subscription URL whatever hub it belonged to, and a single lastEventId meant one hub's cursor could be replayed against another. Both are invisible with one hub and wrong with two. They now live in a per-hub record. A global map still resolves a topic to its hub and matcher, because close(topic) is given nothing else to go on. Two smaller fixes come with it. The subscribe headers are copied before the cursor is written into them: they are the caller's fetch headers, the same object, so writing there leaked Last-Event-Id into every later request. And moving a topic between matchers reconnects once instead of twice, close() being a thin wrapper over a release() that touches no connection. tests/mercure.spec.ts asserted two of its three flags; subscribedToBoth was computed and then dropped. --- packages/mercure/README.md | 2 +- packages/mercure/mercure.ts | 137 ++++++++++++++++++++++-------------- tests/mercure.spec.ts | 2 + 3 files changed, 87 insertions(+), 54 deletions(-) diff --git a/packages/mercure/README.md b/packages/mercure/README.md index 7a75ac8..9808320 100644 --- a/packages/mercure/README.md +++ b/packages/mercure/README.md @@ -88,7 +88,7 @@ When several resources share a matcher, the callbacks passed to the most recent ### Resuming after a disconnection -The id of the last update received is kept and sent back when a subscription is rebuilt, as both the `last_event_id` query parameter and the `Last-Event-Id` request header. The query parameter is what makes this work with a native `EventSource`, which cannot set headers. +The id of the last update received is kept per hub and sent back when a subscription is rebuilt, as both the `last_event_id` query parameter and the `Last-Event-Id` request header. The query parameter is what makes this work with a native `EventSource`, which cannot set headers. ### Examples diff --git a/packages/mercure/mercure.ts b/packages/mercure/mercure.ts index d5df896..fe6d749 100644 --- a/packages/mercure/mercure.ts +++ b/packages/mercure/mercure.ts @@ -25,7 +25,6 @@ type Options = { } & RequestInit; type Subscription = { - mercureUrl: string; type: MatcherType; // The topics this matcher currently stands for. An exact matcher holds one; // a URL Pattern holds every fetched resource it covers, so the subscription @@ -33,20 +32,39 @@ type Subscription = { topics: Set; } -let lastEventId: string -const eventSources = new Map(); -// Matcher (an exact topic, or a URL Pattern) -> the subscription it opens. -const subscriptions = new Map(); -// Topic -> the matcher covering it. -const matchers = new Map(); +// Everything about one hub. Kept per hub rather than in module-wide maps: two +// hubs can legitimately serve the same topic path, and a resume cursor is only +// meaningful to the hub that issued it. +type Hub = { + // Matcher (an exact topic, or a URL Pattern) -> the subscription it opens. + subscriptions: Map; + lastEventId?: string; + eventSource?: any; + options?: Options; +} + +const hubs = new Map() +// Topic -> the hub serving it and the matcher covering it. Global because +// close() is given a topic and nothing else. +const registrations = new Map() + +function hub(mercureUrl: string): Hub { + let entry = hubs.get(mercureUrl) + if (entry === undefined) { + entry = {subscriptions: new Map()} + hubs.set(mercureUrl, entry) + } + + return entry +} // Attach the callbacks to a connection. Split out of listen() so a new // subscriber joining an existing matcher can refresh them without dropping // the stream and reconnecting. -function bind(entry: {eventSource: any, options: Options}, options: Options) { +function bind(entry: Hub, options: Options) { entry.options = options entry.eventSource.onmessage = (event: MessageEvent) => { - lastEventId = event.lastEventId + entry.lastEventId = event.lastEventId if (options.onUpdate) { try { options.onUpdate(options.rawEvent ? event : JSON.parse(event.data)) @@ -60,69 +78,77 @@ function bind(entry: {eventSource: any, options: Options}, options: Option } function listen(mercureUrl: string, options: Options = {}) { - const current = eventSources.get(mercureUrl) - if (current) { - current.eventSource.close() - eventSources.delete(mercureUrl) + const entry = hub(mercureUrl) + if (entry.eventSource) { + entry.eventSource.close() + entry.eventSource = undefined } - const url = new URL(mercureUrl) - let subscribed = 0 - subscriptions.forEach((subscription, matcher) => { - if (subscription.mercureUrl !== mercureUrl) { - return - } + if (entry.subscriptions.size === 0) { + return; + } + const url = new URL(mercureUrl) + entry.subscriptions.forEach((subscription, matcher) => { url.searchParams.append(matcherParam[subscription.type], matcher) - subscribed++ }) - if (subscribed === 0) { - return; - } - - const headers: {[key: string]: string} = options.headers || {} - if (lastEventId) { + // A copy: these headers also belong to the caller's fetch options, and + // writing the cursor into them would leak it into every later request. + const headers: {[key: string]: string} = {...options.headers} + if (entry.lastEventId) { // Every call here opens a fresh connection, so the cursor has to travel // with the request. A native EventSource cannot set headers, hence the // query parameter: the hub takes the union of the query and body // components, and last_event_id is single-valued. The header is sent too, // for EventSource implementations that support it and for the automatic // reconnections they perform on their own. - url.searchParams.append('last_event_id', lastEventId) + url.searchParams.append('last_event_id', entry.lastEventId) // The request header keeps its name in 1.0; only the hub's response header // was renamed to Mercure-Last-Event-ID. - headers['Last-Event-Id'] = lastEventId + headers['Last-Event-Id'] = entry.lastEventId } - const eventSource = new (options.EventSource ?? EventSource)(url.toString(), { withCredentials: options.withCredentials !== undefined ? options.withCredentials : true, headers}); - const entry = {options, eventSource} + entry.eventSource = new (options.EventSource ?? EventSource)(url.toString(), { withCredentials: options.withCredentials !== undefined ? options.withCredentials : true, headers}); bind(entry, options) - eventSources.set(mercureUrl, entry) } -export function close(topic: string) { - const matcher = matchers.get(topic) - if (matcher === undefined) { - return +// Drop a topic from the matcher covering it, without touching any connection. +// Returns the hub whose subscription set changed, so the caller decides when to +// reconnect — moving a topic between matchers changes it twice. +function release(topic: string): string | undefined { + const registration = registrations.get(topic) + if (registration === undefined) { + return undefined } - matchers.delete(topic) + registrations.delete(topic) - const subscription = subscriptions.get(matcher) - if (!subscription) { - return + const entry = hubs.get(registration.mercureUrl) + const subscription = entry?.subscriptions.get(registration.matcher) + if (!entry || !subscription) { + return undefined } subscription.topics.delete(topic) // A URL Pattern covers a family: keep the subscription as long as one of its // topics is still in use. if (subscription.topics.size > 0) { + return undefined + } + + entry.subscriptions.delete(registration.matcher) + + return registration.mercureUrl +} + +export function close(topic: string) { + const mercureUrl = release(topic) + if (mercureUrl === undefined) { return } - subscriptions.delete(matcher) - listen(subscription.mercureUrl, eventSources.get(subscription.mercureUrl)?.options) + listen(mercureUrl, hubs.get(mercureUrl)?.options) } export default async function mercure(url: string, opts: Options) { @@ -151,31 +177,36 @@ export default async function mercure(url: string, opts: Options) { topic = topic === undefined ? url : topic const matcher = opts.matchUrlPattern ?? topic - - // Moving a topic from one matcher to another: release the old one first, - // otherwise it keeps a topic nothing will ever close. - const previous = matchers.get(topic) - if (previous !== undefined && previous !== matcher) { - close(topic) + const entry = hub(mercureUrl) + + // Moving a topic from one matcher to another: drop the old registration + // first, otherwise it keeps a topic nothing will ever close. Released + // rather than closed, so this hub reconnects once below instead of twice. + const previous = registrations.get(topic) + if (previous !== undefined && (previous.matcher !== matcher || previous.mercureUrl !== mercureUrl)) { + const released = release(topic) + // A topic that moved to another hub leaves that one holding a + // subscription it no longer serves. + if (released !== undefined && released !== mercureUrl) { + listen(released, hubs.get(released)?.options) + } } - let subscription = subscriptions.get(matcher) + let subscription = entry.subscriptions.get(matcher) const opened = subscription === undefined if (subscription === undefined) { subscription = { - mercureUrl, type: opts.matchUrlPattern === undefined ? 'exact' : 'urlpattern', topics: new Set(), } - subscriptions.set(matcher, subscription) + entry.subscriptions.set(matcher, subscription) } subscription.topics.add(topic) - matchers.set(topic, matcher) + registrations.set(topic, {mercureUrl, matcher}) - const entry = eventSources.get(mercureUrl) - if (opened || !entry) { + if (opened || !entry.eventSource) { listen(mercureUrl, opts) return res diff --git a/tests/mercure.spec.ts b/tests/mercure.spec.ts index c4e8d90..992c568 100644 --- a/tests/mercure.spec.ts +++ b/tests/mercure.spec.ts @@ -32,6 +32,8 @@ test('mercure', async ({ page }) => { await expect(page.getByTestId('result')).toHaveText('viewing /authors/1: Soyuka'); page.getByTestId('author-2').click({force: true}); await expect(page.getByTestId('result')).toHaveText('viewing /authors/2: O\'Donnell, Peter'); + // Author 1 is still in the tanstack cache, so both topics share one connection. + expect(subscribedToBoth).toBe(true); await page.waitForTimeout(600); // we set gcTime to 500, tanstack query will clear author 1 from cache, therefore we check that mercure gets updated expect(unsubscribedAuthor1).toBe(true); });