feat(server): run playground subscriptions over SSE on SSE-only servers

README.md

@@ -77,6 +77,7 @@ Runnable examples live in [`examples/`](./examples):
 ```bash
 go run ./examples/basic           # queries, mutations, plugins
 go run ./examples/subscriptions   # pub/sub over WebSocket and SSE
+go run ./examples/sse             # SSE-only subscriptions, no WebSocket
 go run ./examples/auth            # bearer auth, context identity, field authorizer
 ```
 

docs/content/docs/examples/index.mdx

@@ -329,6 +329,45 @@ go run ./examples/subscriptions
 
 ---
 
+### Subscriptions over SSE only [#examples-sse]
+
+Registers `sse.New()` as the only subscription transport, so subscriptions are ordinary HTTP requests answered with a `text/event-stream` response - no WebSocket upgrade. The bundled playground detects this and subscribes over SSE too.
+
+#### `examples/sse/main.go`
+
+<include cwd>../examples/sse/main.go</include>
+
+`countdown` produces its own stream and closes it when it reaches zero, which ends the response with a `complete` event. `notifications` relays events published by `sendNotification` through the in-memory bus.
+
+#### `examples/sse/graph/schema.go`
+
+<include cwd>../examples/sse/graph/schema.go</include>
+
+#### `examples/sse/graph/countdown.go`
+
+<include cwd>../examples/sse/graph/countdown.go</include>
+
+#### `examples/sse/graph/notification.go`
+
+<include cwd>../examples/sse/graph/notification.go</include>
+
+```bash
+go run ./examples/sse
+```
+
+Subscribe from a terminal:
+
+```bash
+curl -N http://localhost:4000/graphql \
+  -H 'Content-Type: application/json' \
+  -H 'Accept: text/event-stream' \
+  -d '{"query":"subscription { countdown(from: 3) { remaining } }"}'
+```
+
+**Dig deeper:** [**Realtime subscriptions**](/guides/subscriptions)
+
+---
+
 ## Router integrations [#examples-router-integrations]
 
 `grx` answers **`ServeHTTP`** with exact path matches - you usually pair **`http.StripPrefix("/api", gql)`** with **`WithPlaygroundPath("/playground")`** and **`WithGraphQLPath("/query")`** so callers see **`/api/playground`** and **`POST /api/query`**.

docs/content/docs/guides/subscriptions.mdx

@@ -164,7 +164,9 @@ after the walkthrough.
 
 ## 4. Try it from the playground
 
-The bundled GraphiQL UI auto-detects subscriptions. Run:
+The bundled GraphiQL UI auto-detects subscriptions. It subscribes over
+WebSocket by default and switches to SSE when `sse.New()` is the only
+subscription transport registered. Run:
 
 ```graphql
 subscription {

examples/sse/graph/countdown.go

@@ -0,0 +1,59 @@
+package graph
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+// maxCountdown bounds how long a single countdown stream can stay open.
+const maxCountdown = 60
+
+type Countdown struct {
+	Remaining int `gql:"remaining,nonNull"`
+}
+
+type CountdownArgs struct {
+	From int `gql:"from,nonNull"`
+}
+
+// CountdownSubscription needs no pub/sub: the resolver produces the stream
+// itself and closes the channel when it is done, which ends the SSE response
+// with a "complete" event.
+type CountdownSubscription struct {
+	// Interval is the delay between ticks. Zero means one second.
+	Interval time.Duration
+}
+
+func (s CountdownSubscription) Countdown(ctx context.Context, args CountdownArgs) (<-chan *Countdown, error) {
+	if args.From < 0 || args.From > maxCountdown {
+		return nil, fmt.Errorf("from must be between 0 and %d", maxCountdown)
+	}
+	interval := s.Interval
+	if interval <= 0 {
+		interval = time.Second
+	}
+
+	out := make(chan *Countdown)
+	go func() {
+		defer close(out)
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for remaining := args.From; remaining >= 0; remaining-- {
+			select {
+			case <-ctx.Done():
+				return
+			case out <- &Countdown{Remaining: remaining}:
+			}
+			if remaining == 0 {
+				return
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+		}
+	}()
+	return out, nil
+}

examples/sse/graph/notification.go

@@ -0,0 +1,104 @@
+package graph
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/grx-gql/grx/pubsub"
+)
+
+// notificationTopicPrefix is the bus topic prefix for notification events.
+// Each channel gets its own topic so subscribers only receive its traffic.
+const notificationTopicPrefix = "notification.sent."
+
+func notificationTopic(channel string) string {
+	return notificationTopicPrefix + channel
+}
+
+type Notification struct {
+	ID      string `gql:"id,nonNull"`
+	Channel string `gql:"channel,nonNull"`
+	Text    string `gql:"text,nonNull"`
+	SentAt  string `gql:"sentAt,nonNull"`
+}
+
+type SendNotificationInput struct {
+	Channel string `gql:"channel,nonNull"`
+	Text    string `gql:"text,nonNull"`
+}
+
+type SendNotificationArgs struct {
+	Input SendNotificationInput `gql:"input,nonNull"`
+}
+
+type NotificationsArgs struct {
+	Channel string `gql:"channel,nonNull"`
+}
+
+// NotificationMutation publishes each notification on its channel's topic.
+type NotificationMutation struct {
+	Bus    pubsub.PubSub
+	nextID atomic.Uint64
+}
+
+func (m *NotificationMutation) SendNotification(ctx context.Context, args SendNotificationArgs) (*Notification, error) {
+	channel := strings.TrimSpace(args.Input.Channel)
+	if channel == "" {
+		return nil, fmt.Errorf("channel is required")
+	}
+	if strings.TrimSpace(args.Input.Text) == "" {
+		return nil, fmt.Errorf("text is required")
+	}
+
+	notification := &Notification{
+		ID:      fmt.Sprintf("notification_%d", m.nextID.Add(1)),
+		Channel: channel,
+		Text:    args.Input.Text,
+		SentAt:  time.Now().UTC().Format(time.RFC3339Nano),
+	}
+	payload, err := json.Marshal(notification)
+	if err != nil {
+		return nil, err
+	}
+	if err := m.Bus.Publish(ctx, notificationTopic(channel), payload); err != nil {
+		return nil, err
+	}
+	return notification, nil
+}
+
+// NotificationSubscription relays the notifications published on a channel
+// to every client subscribed to it.
+type NotificationSubscription struct {
+	Bus pubsub.PubSub
+}
+
+func (s NotificationSubscription) Notifications(ctx context.Context, args NotificationsArgs) (<-chan *Notification, error) {
+	channel := strings.TrimSpace(args.Channel)
+	if channel == "" {
+		return nil, fmt.Errorf("channel is required")
+	}
+	msgs, err := s.Bus.Subscribe(ctx, notificationTopic(channel))
+	if err != nil {
+		return nil, err
+	}
+	out := make(chan *Notification)
+	go func() {
+		defer close(out)
+		for msg := range msgs {
+			var notification Notification
+			if err := json.Unmarshal(msg.Payload, &notification); err != nil {
+				continue
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case out <- &notification:
+			}
+		}
+	}()
+	return out, nil
+}

examples/sse/graph/schema.go

@@ -0,0 +1,36 @@
+package graph
+
+import (
+	"context"
+	"time"
+
+	"github.com/grx-gql/grx/pubsub"
+	"github.com/grx-gql/grx/schema"
+)
+
+// Query is the schema's root query type.
+type Query struct{}
+
+// ServerTime returns the server's current UTC time.
+func (Query) ServerTime(ctx context.Context) (string, error) {
+	return time.Now().UTC().Format(time.RFC3339), nil
+}
+
+// Subscription is the schema's root subscription type. Countdown streams on
+// its own; Notifications relays events published by SendNotification.
+type Subscription struct {
+	CountdownSubscription
+	NotificationSubscription
+}
+
+// Schema composes schema.Config. The bus carries notification events from
+// the mutation to every active subscription.
+func Schema(bus pubsub.PubSub) schema.Config {
+	return schema.Config{
+		Query:    Query{},
+		Mutation: &NotificationMutation{Bus: bus},
+		Subscription: Subscription{
+			NotificationSubscription: NotificationSubscription{Bus: bus},
+		},
+	}
+}

examples/sse/graph/schema_test.go

@@ -0,0 +1,100 @@
+package graph_test
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/grx-gql/grx/core"
+	"github.com/grx-gql/grx/examples/sse/graph"
+	"github.com/grx-gql/grx/exec"
+	"github.com/grx-gql/grx/pubsub/memory"
+	"github.com/grx-gql/grx/schema"
+)
+
+func newExecutor(t *testing.T) core.Executor {
+	t.Helper()
+	bus := memory.New()
+	t.Cleanup(func() { _ = bus.Close() })
+	compiled, err := schema.Build(graph.Schema(bus))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return exec.New(compiled, nil)
+}
+
+func TestNotificationsStreamSentNotifications(t *testing.T) {
+	executor := newExecutor(t)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	stream, err := executor.Subscribe(ctx, core.Request{Query: `subscription { notifications(channel: "ops") { text } }`})
+	if err != nil {
+		t.Fatal(err)
+	}
+	response := executor.Execute(ctx, core.Request{Query: `mutation { sendNotification(input: {channel: "ops", text: "deploy done"}) { id } }`})
+	if len(response.Errors) != 0 || response.Data == nil {
+		t.Fatalf("mutation: %+v", response)
+	}
+
+	select {
+	case event, ok := <-stream:
+		if !ok || len(event.Errors) != 0 {
+			t.Fatalf("subscription closed or failed: %+v", event)
+		}
+		encoded, err := json.Marshal(event.Data)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if want := `{"notifications":{"text":"deploy done"}}`; string(encoded) != want {
+			t.Fatalf("got %s, want %s", encoded, want)
+		}
+	case <-ctx.Done():
+		t.Fatal("subscription did not receive the notification")
+	}
+}
+
+func TestCountdownCountsToZeroThenCompletes(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	sub := graph.CountdownSubscription{Interval: time.Millisecond}
+	ticks, err := sub.Countdown(ctx, graph.CountdownArgs{From: 3})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got []int
+	for tick := range ticks {
+		got = append(got, tick.Remaining)
+	}
+	if want := []int{3, 2, 1, 0}; len(got) != len(want) {
+		t.Fatalf("got %v, want %v", got, want)
+	} else {
+		for i := range want {
+			if got[i] != want[i] {
+				t.Fatalf("got %v, want %v", got, want)
+			}
+		}
+	}
+}
+
+func TestCountdownRejectsOutOfRangeStart(t *testing.T) {
+	for _, from := range []int{-1, 61} {
+		if _, err := (graph.CountdownSubscription{}).Countdown(context.Background(), graph.CountdownArgs{From: from}); err == nil {
+			t.Errorf("from=%d: expected an error", from)
+		}
+	}
+}
+
+func TestSendNotificationRequiresChannelAndText(t *testing.T) {
+	executor := newExecutor(t)
+	for _, query := range []string{
+		`mutation { sendNotification(input: {channel: " ", text: "x"}) { id } }`,
+		`mutation { sendNotification(input: {channel: "ops", text: " "}) { id } }`,
+	} {
+		if response := executor.Execute(context.Background(), core.Request{Query: query}); len(response.Errors) == 0 {
+			t.Errorf("%s: expected an error", query)
+		}
+	}
+}