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/README.md b/packages/mercure/README.md index a1e67e7..9808320 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,17 +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 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 -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 1a5e941..fe6d749 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,32 +18,53 @@ 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 = { + 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; +} - if (topics.size === 0) { - return; - } +// 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 url = new URL(mercureUrl) - topics.forEach((_, topic) => { - url.searchParams.append('topic', topic) - }) +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() - const headers: {[key: string]: string} = options.headers || {} - if (lastEventId) { - headers['Last-Event-Id'] = lastEventId +function hub(mercureUrl: string): Hub { + let entry = hubs.get(mercureUrl) + if (entry === undefined) { + entry = {subscriptions: new Map()} + hubs.set(mercureUrl, entry) } - const eventSource = new (options.EventSource ?? EventSource)(url.toString(), { withCredentials: options.withCredentials !== undefined ? options.withCredentials : true, headers}); - eventSource.onmessage = (event: MessageEvent) => { - lastEventId = event.lastEventId + 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: Hub, options: Options) { + entry.options = options + entry.eventSource.onmessage = (event: MessageEvent) => { + entry.lastEventId = event.lastEventId if (options.onUpdate) { try { options.onUpdate(options.rawEvent ? event : JSON.parse(event.data)) @@ -46,22 +74,81 @@ function listen(mercureUrl: string, options: Options = {}) { } } - eventSource.onerror = options.onError - eventSources.set(mercureUrl, { - options: options, - eventSource: eventSource + entry.eventSource.onerror = options.onError +} + +function listen(mercureUrl: string, options: Options = {}) { + const entry = hub(mercureUrl) + if (entry.eventSource) { + entry.eventSource.close() + entry.eventSource = undefined + } + + if (entry.subscriptions.size === 0) { + return; + } + + const url = new URL(mercureUrl) + entry.subscriptions.forEach((subscription, matcher) => { + url.searchParams.append(matcherParam[subscription.type], matcher) }) + + // 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', 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'] = entry.lastEventId + } + + entry.eventSource = new (options.EventSource ?? EventSource)(url.toString(), { withCredentials: options.withCredentials !== undefined ? options.withCredentials : true, headers}); + bind(entry, options) +} + +// 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 + } + + registrations.delete(topic) + + 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) { - if (!topics.has(topic)) { + const mercureUrl = release(topic) + if (mercureUrl === undefined) { return } - const mercureUrl = topics.get(topic) - topics.delete(topic) - const ee = eventSources.get(mercureUrl) - listen(mercureUrl, ee.options) + listen(mercureUrl, hubs.get(mercureUrl)?.options) } export default async function mercure(url: string, opts: Options) { @@ -84,12 +171,53 @@ 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 + 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 = entry.subscriptions.get(matcher) + const opened = subscription === undefined + + if (subscription === undefined) { + subscription = { + type: opts.matchUrlPattern === undefined ? 'exact' : 'urlpattern', + topics: new Set(), + } + entry.subscriptions.set(matcher, subscription) + } + + subscription.topics.add(topic) + registrations.set(topic, {mercureUrl, matcher}) + + if (opened || !entry.eventSource) { 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/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-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-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-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); +}); diff --git a/tests/mercure.spec.ts b/tests/mercure.spec.ts index 84c6829..992c568 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 } }) @@ -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); });