Events Query Modular Implementation

bolt_test.go

@@ -27,8 +27,8 @@ func createBoltTransport(t *testing.T, size uint64, cleanupFrequency float64) *B
 	require.NoError(t, err)
 
 	t.Cleanup(func() {
-		require.NoError(t, os.Remove(path))
 		require.NoError(t, transport.Close(t.Context()))
+		require.NoError(t, os.Remove(path))
 	})
 
 	return transport

caddy/caddy_test.go

@@ -1137,3 +1137,51 @@ func TestLegacyJWTDirectivesRequireExplicitCompatibility(t *testing.T) {
 		})
 	}
 }
+
+func TestEventsQuery(t *testing.T) {
+	tester := caddytest.NewTester(t)
+	tester.InitServer(`
+	{
+		skip_install_trust
+		admin localhost:2999
+		http_port     9080
+		https_port    9443
+	}
+
+	localhost:9080 {
+		route {
+			mercure {
+				anonymous
+				events_query
+				issuer https://example.com {
+					publisher {
+						jwt !ChangeMe!
+					}
+				}
+				resource_identifier https://example.com/.well-known/mercure
+			}
+
+			respond 404
+		}
+	}
+	`, "caddyfile")
+
+	body := url.Values{"match": {"https://example.com/foo/1"}, "events": {""}}
+	req, err := http.NewRequest("QUERY", "http://localhost:9080/.well-known/mercure", strings.NewReader(body.Encode()))
+	require.NoError(t, err)
+	req.Header.Add("Content-Type", "application/x-www-form-urlencoded")
+	req.Header.Add("Events", "duration=1")
+
+	resp := tester.AssertResponseCode(req, http.StatusOK)
+
+	t.Cleanup(func() {
+		require.NoError(t, resp.Body.Close())
+	})
+
+	assert.Equal(t, "application/events+json, application/x-www-form-urlencoded",
+		resp.Header.Get("Accept-Query"))
+	assert.Equal(t, "?1", resp.Header.Get("Incremental"))
+	// The bound the subscription asked for, read from the request and served
+	// back: a hub ignoring it would advertise its own write timeout instead.
+	assert.Equal(t, "duration=1", resp.Header.Get("Events"))
+}

caddy/go.mod

@@ -60,7 +60,9 @@ require (
 	github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da // indirect
 	github.com/dlclark/regexp2/v2 v2.1.1 // indirect
 	github.com/dunglas/go-urlpattern v0.0.0-20260716093037-fb05c4998526 // indirect
+	github.com/dunglas/httpsfv v1.1.1 // indirect
 	github.com/dunglas/skipfilter v1.0.0 // indirect
+	github.com/elnormous/contenttype v1.0.4 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fxamacker/cbor/v2 v2.9.2 // indirect
 	github.com/go-chi/chi/v5 v5.3.0 // indirect

caddy/go.sum

@@ -151,11 +151,15 @@ github.com/dlclark/regexp2/v2 v2.1.1 h1:LCUGyd9Wf+r+VVOl8Ny38JTpWJcAsdVnCIuhhtth
 github.com/dlclark/regexp2/v2 v2.1.1/go.mod h1:avUrQvPaLz2DrFNHJF0taWAFFX2C1GMSSoeiqFjcBmU=
 github.com/dunglas/go-urlpattern v0.0.0-20260716093037-fb05c4998526 h1:biCci7wlx/ChMqOkoUw6FPrhEZYRpulBZ+fO4Geo5Xs=
 github.com/dunglas/go-urlpattern v0.0.0-20260716093037-fb05c4998526/go.mod h1:9qyjDljBPOWyWCGz7vo3Ek7cdnoG/DVk0Ucle7gWVS8=
+github.com/dunglas/httpsfv v1.1.1 h1:HoSs101zIE9I23DlqlmljJ/OIi7ILwrH347pXhRZdxI=
+github.com/dunglas/httpsfv v1.1.1/go.mod h1:zID2mqw9mFsnt7YC3vYQ9/cjq30q41W+1AnDwH8TiMg=
 github.com/dunglas/skipfilter v1.0.0 h1:JG9SgGg4n6BlFwuTYzb9RIqjH7PfwszvWehanrYWPF4=
 github.com/dunglas/skipfilter v1.0.0/go.mod h1:ryhr8j7CAHSjzeN7wI6YEuwoArQ3OQmRqWWVCEAfb9w=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
+github.com/elnormous/contenttype v1.0.4 h1:FjmVNkvQOGqSX70yvocph7keC8DtmJaLzTTq6ZOQCI8=
+github.com/elnormous/contenttype v1.0.4/go.mod h1:5KTOW8m1kdX1dLMiUJeN9szzR2xkngiv2K+RVZwWBbI=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=

caddy/mercure.go

@@ -133,6 +133,9 @@ type Mercure struct {
 	// Dispatch updates when subscriptions are created or terminated
 	Subscriptions bool `json:"subscriptions,omitempty"`
 
+	// Serve subscriptions expressed as an Events Query
+	EventsQuery bool `json:"events_query,omitempty"`
+
 	// Enable the prod-safe debugger UI at /.well-known/mercure/debug/.
 	Debugger bool `json:"debugger,omitempty"`
 
@@ -335,6 +338,10 @@ func (m *Mercure) Provision(ctx caddy.Context) (err error) { //nolint:funlen,goc
 		opts = append(opts, mercure.WithSubscriptions())
 	}
 
+	if m.EventsQuery {
+		opts = append(opts, mercure.WithEventsQuery())
+	}
+
 	if d := m.WriteTimeout; d != nil {
 		opts = append(opts, mercure.WithWriteTimeout(time.Duration(*d)))
 	}
@@ -497,6 +504,9 @@ func (m *Mercure) UnmarshalCaddyfile(d *caddyfile.Dispenser) (err error) { //nol
 			case "subscriptions":
 				m.Subscriptions = true
 
+			case "events_query":
+				m.EventsQuery = true
+
 			case "write_timeout":
 				if m.WriteTimeout, err = parseDurationParameter(d); err != nil {
 					return err

contenttype_test.go

@@ -0,0 +1,120 @@
+package mercure
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestUpdateValidateContentType(t *testing.T) {
+	t.Parallel()
+
+	base := Update{Topics: []string{"https://example.com/books/1"}}
+
+	valid := base
+	valid.ContentType = "application/ld+json; charset=utf-8"
+	require.NoError(t, valid.Validate())
+
+	for _, ct := range []string{"not a media type", "text/plain\r\nX-Injected: 1", "/missing-type"} {
+		invalid := base
+		invalid.ContentType = ct
+		require.ErrorIs(t, invalid.Validate(), ErrInvalidMediaType, ct)
+	}
+}
+
+// History entries persisted before the ContentType field existed must still
+// decode, and the field must survive a marshal/unmarshal round trip.
+func TestUpdateContentTypeJSONRoundTrip(t *testing.T) {
+	t.Parallel()
+
+	u := &Update{
+		Topics:      []string{"https://example.com/books/1"},
+		ContentType: "application/ld+json",
+		Event:       Event{Data: "d", ID: "i"},
+	}
+
+	serialized, err := json.Marshal(u)
+	require.NoError(t, err)
+
+	var decoded Update
+
+	require.NoError(t, json.Unmarshal(serialized, &decoded))
+	assert.Equal(t, *u, decoded)
+
+	var legacy Update
+
+	require.NoError(t, json.Unmarshal([]byte(`{"Topics":["https://example.com/books/1"],"Data":"d","ID":"i"}`), &legacy))
+	assert.Empty(t, legacy.ContentType)
+}
+
+func TestPublishHandlerContentType(t *testing.T) {
+	t.Parallel()
+
+	hub := createDummy(t)
+
+	// The subscriber is registered before the publication, so by the time
+	// PublishHandler returns the update sits in its buffered channel.
+	s := NewLocalSubscriber("", hub.logger, hub.topicMatcherStore)
+	s.SetMatchers([]TopicMatcher{{Type: MatcherTypeExact, Pattern: "https://example.com/books/1"}}, nil)
+	require.NoError(t, hub.transport.AddSubscriber(t.Context(), s))
+
+	form := url.Values{}
+	form.Add("topic", "https://example.com/books/1")
+	form.Add("data", "Hello World")
+	form.Add("content_type", "application/ld+json")
+
+	req := httptest.NewRequest(http.MethodPost, defaultHubURL, strings.NewReader(form.Encode()))
+	req.Header.Add("Content-Type", "application/x-www-form-urlencoded")
+	req.Header.Add("Authorization", bearerPrefix+createDummyAuthorizedJWT(rolePublisher, []string{"*"}))
+
+	w := httptest.NewRecorder()
+	hub.PublishHandler(w, req)
+
+	resp := w.Result()
+
+	t.Cleanup(func() {
+		assert.NoError(t, resp.Body.Close())
+	})
+
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+
+	select {
+	case dispatched := <-s.Receive():
+		assert.Equal(t, "application/ld+json", dispatched.ContentType)
+	case <-time.After(5 * time.Second):
+		t.Fatal("update not received")
+	}
+}
+
+func TestPublishHandlerInvalidContentType(t *testing.T) {
+	t.Parallel()
+
+	hub := createDummy(t)
+
+	form := url.Values{}
+	form.Add("topic", "https://example.com/books/1")
+	form.Add("data", "Hello World")
+	form.Add("content_type", "not a media type")
+
+	req := httptest.NewRequest(http.MethodPost, defaultHubURL, strings.NewReader(form.Encode()))
+	req.Header.Add("Content-Type", "application/x-www-form-urlencoded")
+	req.Header.Add("Authorization", bearerPrefix+createDummyAuthorizedJWT(rolePublisher, []string{"*"}))
+
+	w := httptest.NewRecorder()
+	hub.PublishHandler(w, req)
+
+	resp := w.Result()
+
+	t.Cleanup(func() {
+		assert.NoError(t, resp.Body.Close())
+	})
+
+	assert.Equal(t, http.StatusBadRequest, resp.StatusCode)
+}

encoder.go

@@ -0,0 +1,47 @@
+package mercure
+
+// carrierContentTypes are the media types a stream of notifications can be
+// served as, most preferred first. An Events Query names what it will read in
+// Accept; this list decides what there is to choose from, and what a
+// subscription expressing no preference gets.
+//
+//nolint:gochecknoglobals
+var carrierContentTypes = []string{multipartDigestContentType, eventStreamContentType}
+
+// responseEncoders provides the framing for each media type
+//
+//nolint:gochecknoglobals
+var responseEncoders = map[string]func() streamEncoder{
+	multipartDigestContentType: newMultipartDigestEncoder,
+	eventStreamContentType:     newEventStreamEncoder,
+}
+
+// Mercure ordinarily (without Events Query) serves only Event Stream
+//
+//nolint:gochecknoglobals
+var mercureCarrierContentType = []string{eventStreamContentType}
+
+// streamEncoder frames updates onto a subscription response stream.
+//
+// A subscription response is a sequence of framed messages: a preamble
+// written with the headers, one message per update, and keep-alives in
+// between.
+type streamEncoder interface {
+	// contentType is the response Content-Type field value.
+	contentType() []string
+
+	// preamble is written immediately after the headers and before any
+	// update. Writing it is also what forces the headers onto the wire.
+	preamble() string
+
+	// encode returns the wire form of a single update.
+	encode(u *Update) string
+
+	// heartbeat returns the keep-alive payload.
+	heartbeat() string
+
+	// trailer closes the stream, or "" when the framing needs no terminator.
+	// It is written only when the response ends cleanly, never once the
+	// client has gone away.
+	trailer() string
+}

encoder_eventstream.go

@@ -0,0 +1,24 @@
+package mercure
+
+// The media type a Server-Sent Events stream is served as.
+const eventStreamContentType = "text/event-stream"
+
+// eventStreamEncoder frames updates as Server-Sent Events
+// (text/event-stream), the framing every Mercure subscriber has used since
+// the protocol's first version.
+type eventStreamEncoder struct{}
+
+func newEventStreamEncoder() streamEncoder { return eventStreamEncoder{} }
+
+func (eventStreamEncoder) contentType() []string { return []string{eventStreamContentType} }
+
+// A bare SSE comment. Go currently provides no better way to flush the
+// headers, so writing it is what sends them.
+func (eventStreamEncoder) preamble() string { return ":\n" }
+
+func (eventStreamEncoder) encode(u *Update) string { return u.String() }
+
+// An SSE comment, to prevent issues with some proxies and old browsers.
+func (eventStreamEncoder) heartbeat() string { return ":\n" }
+
+func (eventStreamEncoder) trailer() string { return "" }