Add payload_format raw: send the NATS payload as the event data (#143)

CHANGELOG.md

@@ -8,6 +8,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- `payload_format raw` sends the NATS payload itself as each event's data,
+  without the JSON envelope: a `data:` line per line of it, which
+  EventSource joins back (#143). A payload SSE cannot carry, not UTF-8 text
+  or holding a carriage return, is dropped and counted as
+  `nuts_messages_dropped_total{reason="raw_not_text"}`. The default,
+  `envelope`, changes nothing.
 - `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

Caddyfile.test

@@ -77,6 +77,22 @@
 		}
 	}
 
+	# The same stream with each event's data the raw NATS payload, for
+	# features/sse_streaming.feature.
+	handle /raw* {
+		uri strip_prefix /raw
+		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 *
+			payload_format raw
+		}
+	}
+
 	handle /example* {
 		root * /srv/example
 		uri strip_prefix /example

README.md

@@ -384,6 +384,7 @@ nuts {
     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>
+    payload_format <format>      # Event data: envelope (default) or raw (the NATS payload itself)
     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)
@@ -648,6 +649,30 @@ names themselves, and an event named `error` would reach the page's
 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`.
 
+#### `payload_format`
+
+By default an event's data is the JSON envelope
+`{"topic":…,"payload":…,"time":…}`. `payload_format raw` sends the NATS
+payload itself instead: a page that only wants the payload need not unwrap
+it, plain-text or CSV payloads arrive as they are, and the envelope's 120 to
+150 bytes no longer count towards `max_event_size`.
+
+```
+event: message
+data: first line of the payload
+data: second line of the payload
+id: 12345
+```
+
+A payload with line breaks is sent as several `data:` lines, which
+EventSource joins with `\n` again. SSE data is UTF-8 text and a carriage
+return ends a line, so a payload that is not valid UTF-8 or holds a carriage
+return cannot be sent: it is dropped, logged, and counted as
+`nuts_messages_dropped_total{reason="raw_not_text"}`. The envelope's topic
+and time are gone in raw mode; pair it with
+[`event_type topic`](#event_type) to tell topics apart. The `connected` and
+`reset` events keep their JSON data.
+
 #### `event_id_format`
 
 `event_id_format sequence_time` makes each event's id carry the stored time
@@ -953,7 +978,7 @@ Then scrape `http://localhost:8080/metrics` from Prometheus. Available metrics:
 |--------|------|-------------|
 | `nuts_active_connections` | Gauge | Currently connected SSE clients |
 | `nuts_messages_delivered_total` | Counter | SSE message events successfully written |
-| `nuts_messages_dropped_total{reason}` | Counter (labeled) | Messages not delivered to a client. `reason` is one of `raw_payload` (inbound NATS payload exceeded `max_event_size`), `formatted_sse_message` (SSE envelope after JSON wrap exceeded `max_event_size`), `replay_window` (a replayed message older than `replay_window`) or `control_message` (a subject delete marker or schedule definition; see [Server control messages](#server-control-messages)). |
+| `nuts_messages_dropped_total{reason}` | Counter (labeled) | Messages not delivered to a client. `reason` is one of `raw_payload` (inbound NATS payload exceeded `max_event_size`), `formatted_sse_message` (SSE envelope after JSON wrap exceeded `max_event_size`), `replay_window` (a replayed message older than `replay_window`), `control_message` (a subject delete marker or schedule definition; see [Server control messages](#server-control-messages)) or `raw_not_text` (a payload [`payload_format raw`](#payload_format) cannot send: not UTF-8 text, or holding a carriage return). |
 | `nuts_wildcard_filter_drops_total` | Counter | Deprecated, always 0: the pre-NATS-2.10 wildcard fallback was removed. |
 | `nuts_slow_client_disconnects_total` | Counter | Clients disconnected because a write missed `write_timeout` (the client stopped reading) |
 | `nuts_replay_requests_total` | Counter | Connections requesting message replay |
@@ -1241,7 +1266,7 @@ data: {"topic":"my-topic","payload":{"your":"data"},"time":"2024-01-01T12:00:00Z
 id: 12345
 ```
 
-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 event is named `message` unless [`event_type`](#event_type) names it after its topic or a header, and its data is the JSON envelope unless [`payload_format raw`](#payload_format) sends the payload itself. 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,12 @@ func (h *Handler) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
 				}
 				h.EventIDFormat = d.Val()
 
+			case "payload_format":
+				if !d.NextArg() {
+					return d.ArgErr()
+				}
+				h.PayloadFormat = d.Val()
+
 			case "event_type":
 				if !d.NextArg() {
 					return d.ArgErr()

caddyfile_test.go

@@ -46,6 +46,7 @@ func TestHandler_UnmarshalCaddyfile(t *testing.T) {
 				hub_url https://example.com/events
 				event_id_format sequence_time
 				event_type header Event-Type
+				payload_format raw
 				subscriber_jwt_key secret-key
 				subscriber_jwt_cookie nuts_session
 				allowed_origins https://example.com https://other.com
@@ -81,6 +82,7 @@ func TestHandler_UnmarshalCaddyfile(t *testing.T) {
 				EventIDFormat:             "sequence_time",
 				EventType:                 "header",
 				EventTypeHeader:           "Event-Type",
+				PayloadFormat:             "raw",
 				SubscriberJWTKey:          "secret-key",
 				SubscriberJWTCookie:       "nuts_session",
 				AllowedOrigins:            []string{"https://example.com", "https://other.com"},
@@ -387,6 +389,7 @@ func TestHandler_UnmarshalCaddyfile_MissingArgs(t *testing.T) {
 		{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 payload_format arg", directive: "payload_format"},
 		{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"},
@@ -596,6 +599,12 @@ func TestHandler_UnmarshalCaddyfile_RejectsInvalidOptionalConfig(t *testing.T) {
 			line:    "event_type topic extra",
 			wantErr: "argument",
 		},
+		{
+			name:        "unknown payload_format",
+			line:        "payload_format text",
+			wantErr:     "payload_format must be",
+			validateErr: true,
+		},
 		{
 			name:        "unknown event_type",
 			line:        "event_type category",

docs/CONFIGURATION.md

@@ -76,6 +76,7 @@ enforced. For public or browser-facing routes, issue short-lived tokens with
 | `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`. |
+| `payload_format <format>` | `payload_format` | `envelope` | `envelope` or `raw` | What an event's data holds: the JSON envelope `{"topic","payload","time"}`, or the NATS payload itself, a `data:` line per line of it. In raw mode a payload that is not UTF-8 text, or holds a carriage return, is dropped and counted as `nuts_messages_dropped_total{reason="raw_not_text"}`. |
 | `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

@@ -22,6 +22,14 @@ Feature: SSE Streaming with JetStream
     Then I should receive a "orders" event
     And I should not receive an SSE event with topic "notifications"
 
+  # payload_format raw sends the NATS payload itself as the event's data,
+  # without the JSON envelope (#143).
+  Scenario: Events carry the raw payload
+    Given I am connected to SSE endpoint "/raw?topic=plain"
+    When I publish message 'just some text' to subject "events.plain"
+    Then I should receive an SSE event containing 'just some text'
+    And I should not receive an SSE event containing '"payload"'
+
   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

@@ -8,8 +8,11 @@ import (
 	"strings"
 	"testing"
 	"time"
+	"unicode/utf8"
 
 	"github.com/nats-io/nats.go"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest/observer"
 )
 
 // referenceMessageFrame renders a message frame the way formatMessageEvent
@@ -29,9 +32,19 @@ func referenceMessageFrame(h *Handler, msg streamMessage, now time.Time) string
 	}
 	var event strings.Builder
 	event.WriteString("event: " + referenceEventName(h, msg) + "\n")
-	event.WriteString("data: ")
-	event.WriteString(toJSON(payload))
-	event.WriteString("\n")
+	if h.PayloadFormat == "raw" {
+		text := string(msg.Data)
+		if !utf8.ValidString(text) || strings.ContainsRune(text, '\r') {
+			return ""
+		}
+		for _, line := range strings.Split(text, "\n") {
+			event.WriteString("data: " + line + "\n")
+		}
+	} else {
+		event.WriteString("data: ")
+		event.WriteString(toJSON(payload))
+		event.WriteString("\n")
+	}
 	if msg.HasMetadata {
 		event.WriteString("id: ")
 		event.WriteString(strconv.FormatUint(msg.StreamSequence, 10))
@@ -124,18 +137,24 @@ func TestFormatMessageEvent_MatchesTheReferenceFormatter(t *testing.T) {
 
 func FuzzFormatMessageEventMatchesReference(f *testing.F) {
 	for _, data := range formatterEdgeCases {
-		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(data), "orders", uint64(42), true, false, uint8(0), "", false)
+		f.Add([]byte(data), "orders", uint64(42), true, true, uint8(1), "", false)
+		f.Add([]byte(data), "orders", uint64(42), true, false, uint8(2), "order.created", false)
+		f.Add([]byte(data), "orders", uint64(42), true, false, uint8(0), "", true)
 	}
-	f.Add([]byte(`{}`), "reset", uint64(1), true, false, uint8(1), "")
-	f.Add([]byte(`{}`), "orders", uint64(1), true, false, uint8(2), "has space")
+	f.Add([]byte(`{}`), "reset", uint64(1), true, false, uint8(1), "", false)
+	f.Add([]byte(`{}`), "orders", uint64(1), true, false, uint8(2), "has space", false)
+	f.Add([]byte("line one\nline two\n"), "orders", uint64(1), true, false, uint8(0), "", true)
+	f.Add([]byte("a\r\nb"), "orders", uint64(1), true, false, uint8(0), "", true)
 	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, eventType uint8, header string) {
+	f.Fuzz(func(t *testing.T, data []byte, topic string, seq uint64, hasMeta, timedIDs bool, eventType uint8, header string, raw bool) {
 		h := &Handler{TopicPrefix: "events.", MaxEventSize: -1}
 		if timedIDs {
 			h.EventIDFormat = eventIDSequenceTime
 		}
+		if raw {
+			h.PayloadFormat = payloadFormatRaw
+		}
 		switch eventType % 3 {
 		case 1:
 			h.EventType = eventTypeTopic
@@ -144,12 +163,79 @@ func FuzzFormatMessageEventMatchesReference(f *testing.F) {
 		}
 		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 {
+		got := h.formatMessageEvent(msg, now).Frame
+		if got != want {
 			t.Fatalf("frame differs for data=%q topic=%q:\n got %q\nwant %q", data, topic, got, want)
 		}
+		// #143: what EventSource makes of a raw frame is the payload itself.
+		if raw && got != "" {
+			if back := eventSourceData(got); back != string(data) {
+				t.Fatalf("raw payload %q reads back as %q from %q", data, back, got)
+			}
+		}
 	})
 }
 
+// eventSourceData is the data EventSource dispatches for one frame: the
+// values of its data lines, joined with "\n".
+func eventSourceData(frame string) string {
+	var lines []string
+	for _, line := range strings.Split(strings.TrimSuffix(frame, "\n\n"), "\n") {
+		if value, ok := strings.CutPrefix(line, "data: "); ok {
+			lines = append(lines, value)
+		}
+	}
+	return strings.Join(lines, "\n")
+}
+
+// TestFormatMessageEvent_RawPayload covers #143: in payload_format raw the
+// event's data is the NATS payload, a data line per line of it, which
+// EventSource joins back. A payload SSE cannot carry is dropped and counted;
+// an oversized one keeps its own reason.
+func TestFormatMessageEvent_RawPayload(t *testing.T) {
+	h := &Handler{TopicPrefix: "events.", MaxEventSize: -1, PayloadFormat: payloadFormatRaw}
+	now := time.Date(2026, 9, 27, 12, 0, 0, 0, time.UTC)
+	frame := func(data string) formattedMessageEvent {
+		return h.formatMessageEvent(streamMessage{Subject: "events.orders", Data: []byte(data), HasMetadata: true, StreamSequence: 7, Timestamp: now}, now)
+	}
+	for data, want := range map[string]string{
+		`{"id":1}`:            "event: message\ndata: {\"id\":1}\nid: 7\n\n",
+		"line one\nline two":  "event: message\ndata: line one\ndata: line two\nid: 7\n\n",
+		"ends with a break\n": "event: message\ndata: ends with a break\ndata: \nid: 7\n\n",
+		"":                    "event: message\ndata: \nid: 7\n\n",
+		"zażółć <b>&</b>":     "event: message\ndata: zażółć <b>&</b>\nid: 7\n\n",
+	} {
+		got := frame(data)
+		if got.Dropped || got.Frame != want {
+			t.Errorf("raw %q: frame %q (dropped %v), want %q", data, got.Frame, got.Dropped, want)
+		}
+		if back := eventSourceData(got.Frame); back != data {
+			t.Errorf("raw %q reads back as %q", data, back)
+		}
+	}
+	for _, data := range []string{"a\r\nb", "carriage\rreturn", "\xff\xfe not UTF-8"} {
+		if got := frame(data); !got.Dropped || got.DropReason != dropReasonRawNotText || got.DropSize != len(data) || got.Frame != "" {
+			t.Errorf("raw %q: dropped %v for %q (size %d), want a raw_not_text drop", data, got.Dropped, got.DropReason, got.DropSize)
+		}
+	}
+	h.MaxEventSize = 4
+	if got := frame("\xff too big"); got.DropReason != dropReasonRawPayload {
+		t.Errorf("an oversized payload was dropped for %q, want %q", got.DropReason, dropReasonRawPayload)
+	}
+
+	core, obs := observer.New(zap.WarnLevel)
+	h.logger = zap.New(core)
+	before := metricValue(t, metricsMessagesDropped.WithLabelValues(dropReasonRawNotText))
+	h.MaxEventSize = -1
+	h.recordDroppedMessage(frame("a\rb"))
+	if got := metricValue(t, metricsMessagesDropped.WithLabelValues(dropReasonRawNotText)); got != before+1 {
+		t.Fatalf("messages_dropped_total{raw_not_text} = %v, want %v", got, before+1)
+	}
+	if !hasLogField(obs, "topic", "events.orders") || obs.FilterMessageSnippet("payload_format raw cannot send").Len() != 1 {
+		t.Fatalf("the drop was not logged: %+v", obs.All())
+	}
+}
+
 func BenchmarkFormatMessageEventLarge(b *testing.B) {
 	h := &Handler{TopicPrefix: "events.", MaxEventSize: -1}
 	now := time.Date(2026, 4, 28, 12, 0, 0, 0, time.UTC)

handler.go

@@ -107,6 +107,13 @@ type Handler struct {
 	// EventTypeHeader is the header EventType "header" reads.
 	EventTypeHeader string `json:"event_type_header,omitempty"`
 
+	// PayloadFormat is what an event's data holds: "envelope" (the default),
+	// the JSON {"topic","payload","time"}; or "raw", the NATS payload itself,
+	// one data line per line of it. In raw mode a payload that is not UTF-8
+	// text, or holds a carriage return, cannot be sent as SSE data and is
+	// dropped (#143).
+	PayloadFormat string `json:"payload_format,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.

metrics.go

@@ -39,11 +39,12 @@ var (
 	// producer, the second at envelope overhead on small but pathological
 	// payloads. replay_window counts replayed messages older than the window,
 	// control_message the server's subject delete markers and schedule
-	// definitions.
+	// definitions, raw_not_text the payloads payload_format raw cannot send
+	// (#143).
 	metricsMessagesDropped = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: "nuts",
 		Name:      "messages_dropped_total",
-		Help:      "Total number of messages not delivered to a client, labelled by reason (raw_payload, formatted_sse_message, replay_window, control_message).",
+		Help:      "Total number of messages not delivered to a client, labelled by reason (raw_payload, formatted_sse_message, replay_window, control_message, raw_not_text).",
 	}, []string{"reason"})
 
 	// nuts_wildcard_filter_drops_total is deprecated: it counted messages the