Add event_type: name SSE events after their topic or a header (#142)

CHANGELOG.md

@@ -8,6 +8,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- `event_type topic` and `event_type header <name>` name each message's SSE
+  event after its topic or a header of the NATS message, so a page can add
+  one `addEventListener` per kind of event instead of routing every
+  `message` itself (#142). Names outside `[A-Za-z0-9._:-]{1,100}`, the
+  reserved `connected`, `reset`, `open` and `error`, and messages without the
+  header fall back to `message`, the default.
 - `event_id_format sequence_time` makes each event's id carry the stored time
   of its message with its stream sequence, `<sequence>-<unix ns>` (#138). A
   client that was away while its stream was recreated or restored comes back

Caddyfile.test

@@ -61,6 +61,22 @@
 		}
 	}
 
+	# The same stream with each event named after its topic, for
+	# features/sse_streaming.feature.
+	handle /typed* {
+		uri strip_prefix /typed
+		nuts {
+			nats_url {$NATS_URL:nats://localhost:4222}
+			stream_name {$STREAM_NAME:EVENTS}
+			topic_prefix {$TOPIC_PREFIX:events.}
+			heartbeat_interval 1
+			nats_idle_heartbeat 1
+			nats_ping_interval 1
+			allowed_origins *
+			event_type topic
+		}
+	}
+
 	handle /example* {
 		root * /srv/example
 		uri strip_prefix /example

README.md

@@ -383,6 +383,7 @@ nuts {
     replay_max_messages <count>  # Cap replayed messages per reconnect (default: 0 = unlimited)
     replay_window <seconds>      # Time-bound replay to the last N seconds (default: 0 = all retained)
     event_id_format <format>     # Event ids: sequence (default) or sequence_time (sequence and message time)
+    event_type <source> [name]   # Event names: message (default), topic, or header <name>
     health_path <path>           # Legacy readiness endpoint (empty/default: /healthz)
     live_path <path>             # Process liveness endpoint (empty/default: /livez)
     ready_path <path>            # NATS/stream readiness endpoint (empty/default: /readyz)
@@ -627,6 +628,26 @@ streams. Pick bounds that match the largest replay you are willing to serve to
 one client, then size JetStream retention, `max_connections`, and edge
 rate-limits around that budget.
 
+#### `event_type`
+
+Every message is sent as `event: message` by default, with its topic in the
+JSON data, so a page handles all of them in one `message` listener.
+`event_type` names each event after something of the message instead:
+
+- `event_type topic`: the topic, without `topic_prefix`. A page subscribed to
+  several topics can add one listener per topic:
+  `events.addEventListener('orders', …)`.
+- `event_type header <name>`: the value of that header of the NATS message,
+  for example `event_type header Event-Type` with publishers that set
+  `Event-Type: order_created`. Header names are case-sensitive in NATS.
+
+A name must be 1 to 100 characters of `A-Z a-z 0-9 . _ : -`, and it must not
+be `connected`, `reset`, `open` or `error`: NUTS and EventSource use those
+names themselves, and an event named `error` would reach the page's
+`onerror` handler as if the connection had failed. Otherwise, and for a
+message without the header, the event is named `message`. The JSON data is
+the same in every case. The event name counts towards `max_event_size`.
+
 #### `event_id_format`
 
 `event_id_format sequence_time` makes each event's id carry the stored time
@@ -1220,7 +1241,7 @@ data: {"topic":"my-topic","payload":{"your":"data"},"time":"2024-01-01T12:00:00Z
 id: 12345
 ```
 
-The `id` field contains the JetStream sequence number, which can be used with `last-id` or `Last-Event-ID` for replay. With [`event_id_format sequence_time`](#event_id_format) it also carries the stored time of the message: `id: 12345-1790513389663997004`. It is the last line of each event: SSE allows the fields in any order, and some clients (for example `@microsoft/fetch-event-source`) record an id as soon as they read it, so with the id last a frame cut off mid-write cannot make them resume after a message they never received.
+The event is named `message` unless [`event_type`](#event_type) names it after its topic or a header. The `id` field contains the JetStream sequence number, which can be used with `last-id` or `Last-Event-ID` for replay. With [`event_id_format sequence_time`](#event_id_format) it also carries the stored time of the message: `id: 12345-1790513389663997004`. It is the last line of each event: SSE allows the fields in any order, and some clients (for example `@microsoft/fetch-event-source`) record an id as soon as they read it, so with the id last a frame cut off mid-write cannot make them resume after a message they never received.
 
 The first frame of every stream is the handshake event:
 

caddyfile.go

@@ -302,6 +302,21 @@ func (h *Handler) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
 				}
 				h.EventIDFormat = d.Val()
 
+			case "event_type":
+				if !d.NextArg() {
+					return d.ArgErr()
+				}
+				h.EventType = d.Val()
+				if h.EventType == eventTypeHeader {
+					if !d.NextArg() {
+						return d.Errf("event_type header needs the header's name")
+					}
+					h.EventTypeHeader = d.Val()
+				}
+				if d.NextArg() {
+					return d.ArgErr()
+				}
+
 			default:
 				return d.Errf("unrecognized option: %s", d.Val())
 			}

caddyfile_test.go

@@ -45,6 +45,7 @@ func TestHandler_UnmarshalCaddyfile(t *testing.T) {
 				ready_path /ready
 				hub_url https://example.com/events
 				event_id_format sequence_time
+				event_type header Event-Type
 				subscriber_jwt_key secret-key
 				subscriber_jwt_cookie nuts_session
 				allowed_origins https://example.com https://other.com
@@ -78,6 +79,8 @@ func TestHandler_UnmarshalCaddyfile(t *testing.T) {
 				ReadyPath:                 "/ready",
 				HubURL:                    "https://example.com/events",
 				EventIDFormat:             "sequence_time",
+				EventType:                 "header",
+				EventTypeHeader:           "Event-Type",
 				SubscriberJWTKey:          "secret-key",
 				SubscriberJWTCookie:       "nuts_session",
 				AllowedOrigins:            []string{"https://example.com", "https://other.com"},
@@ -383,6 +386,7 @@ func TestHandler_UnmarshalCaddyfile_MissingArgs(t *testing.T) {
 		{name: "missing reconnect_wait arg", directive: "reconnect_wait"},
 		{name: "missing nats_ping_interval arg", directive: "nats_ping_interval"},
 		{name: "missing event_id_format arg", directive: "event_id_format"},
+		{name: "missing event_type arg", directive: "event_type"},
 		{name: "missing max_reconnects arg", directive: "max_reconnects"},
 		{name: "missing max_event_size arg", directive: "max_event_size"},
 		{name: "missing dispatch_timeout arg", directive: "dispatch_timeout"},
@@ -582,6 +586,28 @@ func TestHandler_UnmarshalCaddyfile_RejectsInvalidOptionalConfig(t *testing.T) {
 			line:    "nats_ping_interval -5",
 			wantErr: "nats_ping_interval must be >= 0",
 		},
+		{
+			name:    "event_type header without a name",
+			line:    "event_type header",
+			wantErr: "event_type header needs the header's name",
+		},
+		{
+			name:    "event_type with an extra argument",
+			line:    "event_type topic extra",
+			wantErr: "argument",
+		},
+		{
+			name:        "unknown event_type",
+			line:        "event_type category",
+			wantErr:     "event_type must be",
+			validateErr: true,
+		},
+		{
+			name:        "event_type header with a colon in the name",
+			line:        "event_type header Event:Type",
+			wantErr:     "header name",
+			validateErr: true,
+		},
 		{
 			name:    "topic cap below the -1 sentinel",
 			line:    "max_topics_per_subscription -2",

docs/CONFIGURATION.md

@@ -74,6 +74,8 @@ enforced. For public or browser-facing routes, issue short-lived tokens with
 | `dispatch_timeout <seconds>` | `dispatch_timeout` | `0` | Integer `>= 0` | Deprecated, no effect: the pull consumer has no queue hand-off to time out. A positive value logs a warning. It will be removed in the next major release. |
 | `write_timeout <seconds>` | `write_timeout` | `30` when `0` or omitted | Integer `>= 0`, or `-1` to disable | Deadline for each SSE write and flush. A client that stops reading is disconnected when a write misses it (`disconnect_reason=slow_client`) and resumes with its last event ID. `-1` leaves write deadlines to Caddy's server config. |
 | `replay_max_messages <count>` | `replay_max_messages` | `0` | Integer `>= 0` | `0` is unlimited. When reached during replay, NUTS closes the stream cleanly and the client reconnects from its last event ID. Only history counts: messages published after the request arrived never trip the cap. |
+| `event_type <source> [header name]` | `event_type` | `message` | `message`, `topic`, or `header` followed by a header name | The SSE event name of each message: `message`, its topic without `topic_prefix`, or the value of a NATS message header. A name outside `[A-Za-z0-9._:-]{1,100}`, a reserved name (`connected`, `reset`, `open`, `error`) or a missing header falls back to `message`. |
+| (the name after `event_type header`) | `event_type_header` | Empty | Printable ASCII without spaces or colons | The header `event_type header` reads; case-sensitive, as NATS headers are. Rejected unless `event_type` is `header`. |
 | `event_id_format <format>` | `event_id_format` | `sequence` | `sequence` or `sequence_time` | What an event's id holds: the message's stream sequence, or the sequence and the stored time of the message, `<sequence>-<unix ns>`. With `sequence_time`, a client that resumes with a cursor from a stream that was since recreated or restored replays the stream from its start instead of skipping its first messages; both forms are accepted as cursors either way. Enable it only for clients that treat ids as opaque strings. |
 | `replay_window <seconds>` | `replay_window` | `0` | Integer `>= 0` | `0` preserves retained replay. Positive values bound old replay cursors to `StartTime(now - replay_window)` while preserving exact sequence replay inside the window. Also applies when stream info cannot be read or the cursor is ahead of the stream. Live messages are never filtered. |
 

features/sse_streaming.feature

@@ -14,6 +14,14 @@ Feature: SSE Streaming with JetStream
     And the event payload should contain "alert"
     And the event should have an ID
 
+  # event_type topic names each event after its topic (#142), so a page can
+  # add one listener per topic instead of routing every message itself.
+  Scenario: Events are named after their topic
+    Given I am connected to SSE endpoint "/typed?topic=orders"
+    When I publish message '{"id": 1}' to subject "events.orders"
+    Then I should receive a "orders" event
+    And I should not receive an SSE event with topic "notifications"
+
   Scenario: Receive messages from multiple topics
     Given I am connected to SSE endpoint "/events?topic=alerts&topic=updates"
     When I publish message '{"type": "alert"}' to subject "events.alerts"

formatter_test.go

@@ -3,10 +3,13 @@ package nuts
 import (
 	"bytes"
 	"encoding/json"
+	"regexp"
 	"strconv"
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/nats-io/nats.go"
 )
 
 // referenceMessageFrame renders a message frame the way formatMessageEvent
@@ -25,7 +28,7 @@ func referenceMessageFrame(h *Handler, msg streamMessage, now time.Time) string
 		payload.Time = now.UTC().Format(time.RFC3339)
 	}
 	var event strings.Builder
-	event.WriteString("event: message\n")
+	event.WriteString("event: " + referenceEventName(h, msg) + "\n")
 	event.WriteString("data: ")
 	event.WriteString(toJSON(payload))
 	event.WriteString("\n")
@@ -41,6 +44,27 @@ func referenceMessageFrame(h *Handler, msg streamMessage, now time.Time) string
 	return event.String()
 }
 
+// referenceEventName is the reference formatter's event_type step, written
+// with a regular expression rather than the handler's byte loop.
+var referenceEventNamePattern = regexp.MustCompile(`^[A-Za-z0-9._:-]{1,100}$`)
+
+func referenceEventName(h *Handler, msg streamMessage) string {
+	var name string
+	switch h.EventType {
+	case "topic":
+		name = strings.TrimPrefix(msg.Subject, h.TopicPrefix)
+	case "header":
+		name = msg.Header.Get(h.EventTypeHeader)
+	default:
+		return "message"
+	}
+	switch {
+	case !referenceEventNamePattern.MatchString(name), name == "connected", name == "reset", name == "open", name == "error":
+		return "message"
+	}
+	return name
+}
+
 // tryParseJSON is the reference formatter's payload step: valid JSON is
 // compacted into a json.RawMessage, anything else is kept as a string.
 func tryParseJSON(data []byte) any {
@@ -100,16 +124,25 @@ func TestFormatMessageEvent_MatchesTheReferenceFormatter(t *testing.T) {
 
 func FuzzFormatMessageEventMatchesReference(f *testing.F) {
 	for _, data := range formatterEdgeCases {
-		f.Add([]byte(data), "orders", uint64(42), true, false)
-		f.Add([]byte(data), "orders", uint64(42), true, true)
+		f.Add([]byte(data), "orders", uint64(42), true, false, uint8(0), "")
+		f.Add([]byte(data), "orders", uint64(42), true, true, uint8(1), "")
+		f.Add([]byte(data), "orders", uint64(42), true, false, uint8(2), "order.created")
 	}
+	f.Add([]byte(`{}`), "reset", uint64(1), true, false, uint8(1), "")
+	f.Add([]byte(`{}`), "orders", uint64(1), true, false, uint8(2), "has space")
 	now := time.Date(2026, 9, 26, 12, 0, 0, 0, time.UTC)
-	f.Fuzz(func(t *testing.T, data []byte, topic string, seq uint64, hasMeta, timedIDs bool) {
+	f.Fuzz(func(t *testing.T, data []byte, topic string, seq uint64, hasMeta, timedIDs bool, eventType uint8, header string) {
 		h := &Handler{TopicPrefix: "events.", MaxEventSize: -1}
 		if timedIDs {
 			h.EventIDFormat = eventIDSequenceTime
 		}
-		msg := streamMessage{Subject: "events." + topic, Data: data, HasMetadata: hasMeta, StreamSequence: seq, Timestamp: now.Add(-time.Minute)}
+		switch eventType % 3 {
+		case 1:
+			h.EventType = eventTypeTopic
+		case 2:
+			h.EventType, h.EventTypeHeader = eventTypeHeader, "Event-Type"
+		}
+		msg := streamMessage{Subject: "events." + topic, Data: data, Header: nats.Header{"Event-Type": []string{header}}, HasMetadata: hasMeta, StreamSequence: seq, Timestamp: now.Add(-time.Minute)}
 		want := referenceMessageFrame(h, msg, now)
 		if got := h.formatMessageEvent(msg, now).Frame; got != want {
 			t.Fatalf("frame differs for data=%q topic=%q:\n got %q\nwant %q", data, topic, got, want)
@@ -196,3 +229,44 @@ func BenchmarkWriteJSONPayload(b *testing.B) {
 		})
 	}
 }
+
+// TestHandler_EventName: event_type takes a message's SSE event name from
+// its topic or a header, and falls back to "message" when that gives no
+// usable name (#142).
+func TestHandler_EventName(t *testing.T) {
+	msg := func(subject, header string) streamMessage {
+		m := streamMessage{Subject: subject}
+		if header != "-" {
+			m.Header = nats.Header{"Event-Type": []string{header}}
+		}
+		return m
+	}
+	topic := &Handler{TopicPrefix: "events.", EventType: eventTypeTopic}
+	header := &Handler{TopicPrefix: "events.", EventType: eventTypeHeader, EventTypeHeader: "Event-Type"}
+	for _, c := range []struct {
+		name string
+		h    *Handler
+		msg  streamMessage
+		want string
+	}{
+		{"default", &Handler{TopicPrefix: "events."}, msg("events.orders", "created"), "message"},
+		{"message", &Handler{TopicPrefix: "events.", EventType: eventTypeMessage}, msg("events.orders", "created"), "message"},
+		{"topic", topic, msg("events.orders.created", "x"), "orders.created"},
+		{"topic without the prefix", topic, msg("other.orders", "x"), "other.orders"},
+		{"topic that is reserved", topic, msg("events.error", "x"), "message"},
+		{"header", header, msg("events.orders", "order:created"), "order:created"},
+		{"header missing", header, msg("events.orders", "-"), "message"},
+		{"header empty", header, msg("events.orders", ""), "message"},
+		{"header with a line break", header, msg("events.orders", "a\nevent: forged"), "message"},
+		{"header reserved", header, msg("events.orders", "connected"), "message"},
+		{"header of another case", &Handler{EventType: eventTypeHeader, EventTypeHeader: "event-type"}, msg("events.orders", "created"), "message"},
+	} {
+		if got := c.h.eventName(c.msg); got != c.want {
+			t.Errorf("%s: eventName = %q, want %q", c.name, got, c.want)
+		}
+	}
+	frame := topic.formatMessageEvent(streamMessage{Subject: "events.orders", Data: []byte(`{}`)}, time.Date(2026, 9, 27, 0, 0, 0, 0, time.UTC)).Frame
+	if !strings.HasPrefix(frame, "event: orders\ndata: ") {
+		t.Fatalf("frame = %q, want the orders event", frame)
+	}
+}

handler.go

@@ -98,6 +98,15 @@ type Handler struct {
 	// recreated or restored replays the stream from its start (#138).
 	EventIDFormat string `json:"event_id_format,omitempty"`
 
+	// EventType sets the SSE event name of each message: "message" (the
+	// default); "topic", the message's topic without topic_prefix; or
+	// "header", the value of the message header EventTypeHeader names. A name
+	// that is empty, reserved (connected, reset, open, error) or not made of
+	// [A-Za-z0-9._:-] falls back to "message" (#142).
+	EventType string `json:"event_type,omitempty"`
+	// EventTypeHeader is the header EventType "header" reads.
+	EventTypeHeader string `json:"event_type_header,omitempty"`
+
 	// HubURL is the URL advertised in the Link header for hub discovery.
 	// When set, SSE responses include a Link: <url>; rel="nuts" header
 	// so that clients and upstream APIs can discover the event hub automatically.

helpers.go

@@ -258,6 +258,43 @@ func isAllowedTopicByte(c byte) bool {
 		c == '.' || c == '-' || c == '_'
 }
 
+// maxEventNameLen bounds an event name taken from a topic or a header.
+const maxEventNameLen = 100
+
+// isUsableEventName reports whether name can be a message's SSE event name
+// (#142): 1 to maxEventNameLen bytes of [A-Za-z0-9._:-], and not a name NUTS
+// or EventSource uses for something else. An event named error would reach
+// the page's onerror handler as if the connection had failed.
+func isUsableEventName(name string) bool {
+	if name == "" || len(name) > maxEventNameLen {
+		return false
+	}
+	switch name {
+	case "connected", "reset", "open", "error":
+		return false
+	}
+	for i := 0; i < len(name); i++ {
+		if c := name[i]; !isAllowedTopicByte(c) && c != ':' {
+			return false
+		}
+	}
+	return true
+}
+
+// isValidHeaderName accepts a NATS header name: 1 to 256 bytes of printable
+// ASCII without spaces or colons.
+func isValidHeaderName(name string) bool {
+	if name == "" || len(name) > 256 {
+		return false
+	}
+	for i := 0; i < len(name); i++ {
+		if c := name[i]; c <= ' ' || c > '~' || c == ':' {
+			return false
+		}
+	}
+	return true
+}
+
 // isValidTopic rejects topic names that would be problematic as NATS
 // subjects. Accepted character set: ASCII letters, digits, dot, dash,
 // underscore. Rejects wildcards (* and >), the system prefix ($),