From d039c9e535e14cfeb0479ee81194c9713e37860a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bartosz=20Pacho=C5=82ek?= Date: Sun, 27 Sep 2026 23:06:57 +0200 Subject: [PATCH] Add event_type: name SSE events after their topic or a header (#142) Every message went out as `event: message`, with its topic inside the JSON data, so a page handled all of them in one listener and routed them itself. event_type topic names each event after its topic, without topic_prefix. event_type header uses that header of the NATS message. A name must be 1 to 100 bytes of [A-Za-z0-9._:-] and must not be one NUTS or EventSource uses itself (connected, reset, open, error): an event named error would reach the page's onerror handler. Otherwise, and for a message without the header, the event keeps the name message. The default, message, changes nothing. Closes #142 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 6 +++ Caddyfile.test | 16 +++++++ README.md | 23 +++++++++- caddyfile.go | 15 ++++++ caddyfile_test.go | 26 +++++++++++ docs/CONFIGURATION.md | 2 + features/sse_streaming.feature | 8 ++++ formatter_test.go | 84 ++++++++++++++++++++++++++++++++-- handler.go | 9 ++++ helpers.go | 37 +++++++++++++++ helpers_test.go | 48 +++++++++++++++++++ provision.go | 15 ++++++ provision_test.go | 15 ++++++ serve.go | 28 +++++++++++- serve_integration_test.go | 61 ++++++++++++++++++++++++ website/docs/usage.md | 1 + 16 files changed, 387 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 99453a6..c315a94 100644 --- a/CHANGELOG.md +++ b/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 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, `-` (#138). A client that was away while its stream was recreated or restored comes back diff --git a/Caddyfile.test b/Caddyfile.test index 30fc0a2..b274151 100644 --- a/Caddyfile.test +++ b/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 diff --git a/README.md b/README.md index 2caac84..a7f436f 100644 --- a/README.md +++ b/README.md @@ -383,6 +383,7 @@ nuts { replay_max_messages # Cap replayed messages per reconnect (default: 0 = unlimited) replay_window # Time-bound replay to the last N seconds (default: 0 = all retained) event_id_format # Event ids: sequence (default) or sequence_time (sequence and message time) + event_type [name] # Event names: message (default), topic, or header health_path # Legacy readiness endpoint (empty/default: /healthz) live_path # Process liveness endpoint (empty/default: /livez) ready_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 `: 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: diff --git a/caddyfile.go b/caddyfile.go index 7047bd7..a366275 100644 --- a/caddyfile.go +++ b/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()) } diff --git a/caddyfile_test.go b/caddyfile_test.go index 32fe4d5..0cd6e27 100644 --- a/caddyfile_test.go +++ b/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", diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 95a0822..e138d98 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -74,6 +74,8 @@ enforced. For public or browser-facing routes, issue short-lived tokens with | `dispatch_timeout ` | `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 ` | `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 ` | `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 [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 ` | `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, `-`. 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 ` | `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. | diff --git a/features/sse_streaming.feature b/features/sse_streaming.feature index 6cec1b2..c348260 100644 --- a/features/sse_streaming.feature +++ b/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" diff --git a/formatter_test.go b/formatter_test.go index 0dbde53..3c56803 100644 --- a/formatter_test.go +++ b/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) + } +} diff --git a/handler.go b/handler.go index 3f9de07..b960d03 100644 --- a/handler.go +++ b/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: ; rel="nuts" header // so that clients and upstream APIs can discover the event hub automatically. diff --git a/helpers.go b/helpers.go index 345f09d..86efce0 100644 --- a/helpers.go +++ b/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 ($), diff --git a/helpers_test.go b/helpers_test.go index 0b41815..22dced9 100644 --- a/helpers_test.go +++ b/helpers_test.go @@ -396,3 +396,51 @@ func TestWriteSSEChunkWithTimeout_SetsAndClearsDeadline(t *testing.T) { t.Fatalf("second deadline = %v, want zero reset", rr.deadlines[1]) } } + +// TestIsUsableEventName: a message's SSE event name taken from a topic or a +// header must be a plain token that no one else uses (#142). +func TestIsUsableEventName(t *testing.T) { + for name, want := range map[string]bool{ + "orders": true, + "orders.created": true, + "Order_Created-v2:eu": true, + strings.Repeat("a", 100): true, + strings.Repeat("a", 101): false, + "": false, + "connected": false, + "reset": false, + "open": false, + "error": false, + "message": true, + "has space": false, + "line\nbreak": false, + "carriage\rreturn": false, + "slash/and$dollar": false, + "zażółć": false, + "tab\there": false, + } { + if got := isUsableEventName(name); got != want { + t.Errorf("isUsableEventName(%q) = %v, want %v", name, got, want) + } + } +} + +func TestIsValidHeaderName(t *testing.T) { + for name, want := range map[string]bool{ + "Event-Type": true, + "Nats-Msg-Id": true, + "x_custom.type!": true, + strings.Repeat("h", 256): true, + strings.Repeat("h", 257): false, + "": false, + "has space": false, + "colon:inside": false, + "tab\t": false, + "del\x7f": false, + "non-ascii-ł": false, + } { + if got := isValidHeaderName(name); got != want { + t.Errorf("isValidHeaderName(%q) = %v, want %v", name, got, want) + } + } +} diff --git a/provision.go b/provision.go index fd96b5e..2947dd1 100644 --- a/provision.go +++ b/provision.go @@ -210,6 +210,9 @@ func (h *Handler) Provision(ctx caddy.Context) error { if h.EventIDFormat == "" { h.EventIDFormat = eventIDSequence } + if h.EventType == "" { + h.EventType = eventTypeMessage + } if h.HealthPath == "" { h.HealthPath = defaultHealthPath } @@ -864,6 +867,18 @@ func (h *Handler) validateConfigValues() error { default: return fmt.Errorf("event_id_format must be %q or %q, not %q", eventIDSequence, eventIDSequenceTime, h.EventIDFormat) } + switch h.EventType { + case "", eventTypeMessage, eventTypeTopic: + if h.EventTypeHeader != "" { + return fmt.Errorf("event_type_header is only read with event_type header") + } + case eventTypeHeader: + if !isValidHeaderName(h.EventTypeHeader) { + return fmt.Errorf("event_type header needs a header name of printable ASCII without spaces or colons, not %q", h.EventTypeHeader) + } + default: + return fmt.Errorf("event_type must be %q, %q or %q, not %q", eventTypeMessage, eventTypeTopic, eventTypeHeader, h.EventType) + } if h.MaxTopicsPerSubscription < maxTopicsDisabledSentinel { return fmt.Errorf("max_topics_per_subscription (%d) is invalid: the only accepted negative value is -1 (no limit); other negatives are rejected as typos", h.MaxTopicsPerSubscription) diff --git a/provision_test.go b/provision_test.go index b0f6dbf..545bdfa 100644 --- a/provision_test.go +++ b/provision_test.go @@ -1189,6 +1189,21 @@ func TestHandler_Provision_RejectsInvalidOptionalJSONConfigBeforeDialing(t *test fragment: `"event_id_format": "timestamp"`, wantErr: "event_id_format", }, + { + name: "unknown event type", + fragment: `"event_type": "category"`, + wantErr: "event_type must be", + }, + { + name: "event type header without header mode", + fragment: `"event_type": "topic", "event_type_header": "Event-Type"`, + wantErr: "event_type_header", + }, + { + name: "event type header mode without a name", + fragment: `"event_type": "header"`, + wantErr: "header name", + }, { name: "nats_idle_heartbeat at boundary equals InactiveThreshold/2", fragment: `"nats_idle_heartbeat": 15`, diff --git a/serve.go b/serve.go index e30ad61..fe987d7 100644 --- a/serve.go +++ b/serve.go @@ -72,6 +72,12 @@ const ( // nanoseconds, "-