feat(mercure): subscribe with a URL Pattern

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
 }

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

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: <https://localhost/authors/1>; rel="self"
 Link: <https://localhost/.well-known/mercure>; 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).

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<MatcherType, string> = {
+  exact: 'match',
+  urlpattern: 'match_urlpattern',
+}
 
 type Options<T> = {
   rawEvent?: boolean;
@@ -11,32 +18,53 @@ type Options<T> = {
   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<T>(mercureUrl: string, options: Options<T> = {}) {
-  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<string>;
+}
 
-  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<string, Subscription>;
+  lastEventId?: string;
+  eventSource?: any;
+  options?: Options<any>;
+}
 
-  const url = new URL(mercureUrl)
-  topics.forEach((_, topic) => {
-    url.searchParams.append('topic', topic)
-  })
+const hubs = new Map<string, Hub>()
+// Topic -> the hub serving it and the matcher covering it. Global because
+// close() is given a topic and nothing else.
+const registrations = new Map<string, {mercureUrl: string, matcher: string}>()
 
-  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<string, Subscription>()}
+    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<T>(entry: Hub, options: Options<T>) {
+  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<T>(mercureUrl: string, options: Options<T> = {}) {
     }
   }
 
-  eventSource.onerror = options.onError
-  eventSources.set(mercureUrl, {
-    options: options,
-    eventSource: eventSource
+  entry.eventSource.onerror = options.onError
+}
+
+function listen<T>(mercureUrl: string, options: Options<T> = {}) {
+  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<T>(url: string, opts: Options<T>) {
@@ -84,12 +171,53 @@ export default async function mercure<T>(url: string, opts: Options<T>) {
           }
         });
 
-      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<string>(),
+        }
+        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;
     });
 }
-

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'
        }
      });
     }