Merge pull request #17 from galaxy-io/atterpac/browse-bugs

browse bugs

Author: atterpac

cmd/seeder/seeder.go

@@ -99,8 +99,11 @@ func seedStreams(ctx context.Context, js jetstream.JetStream) {
 			Discard:     jetstream.DiscardOld,
 		},
 		{
-			Name:        "EVENTS",
-			Subjects:    []string{"events.>"},
+			Name: "EVENTS",
+			// Multiple disjoint subjects: "audit.>" does NOT start with
+			// "events." — watching the stream must surface these too, not just
+			// the first configured subject.
+			Subjects:    []string{"events.>", "audit.>"},
 			Retention:   jetstream.InterestPolicy,
 			MaxAge:      72 * time.Hour,
 			Storage:     jetstream.FileStorage,
@@ -329,6 +332,18 @@ func seedMessages(ctx context.Context, js jetstream.JetStream, nc *nats.Conn) {
 		})
 	}
 
+	// Audit — second subject on the EVENTS stream that does NOT start with
+	// "events.". Watching the EVENTS stream should surface these alongside
+	// the events.* messages; the old single-subject watch missed them.
+	auditActions := []string{"login", "logout", "permission.grant", "permission.revoke", "config.change"}
+	for i := 0; i < 80; i++ {
+		action := auditActions[rand.Intn(len(auditActions))]
+		messages = append(messages, msg{
+			subject: fmt.Sprintf("audit.%s", action),
+			data:    fmt.Sprintf(`{"action":"%s","actor":"admin-%03d","target":"usr-%05d","ip":"10.0.%d.%d","result":"%s"}`, action, rand.Intn(50), rand.Intn(10000), rand.Intn(256), rand.Intn(256), []string{"allow", "deny"}[rand.Intn(2)]),
+		})
+	}
+
 	// Logs
 	for i := 0; i < 500; i++ {
 		level := levels[rand.Intn(len(levels))]

go.mod

@@ -3,7 +3,7 @@ module github.com/galaxy-io/gnat
 go 1.25.0
 
 require (
-	github.com/atterpac/dado v0.2.9
+	github.com/atterpac/dado v0.2.10
 	github.com/gdamore/tcell/v2 v2.13.4
 	github.com/nats-io/nats.go v1.39.1
 	golang.org/x/term v0.37.0

go.sum

@@ -1,5 +1,5 @@
-github.com/atterpac/dado v0.2.9 h1:wMDErpS1oMKC8J7NJd53oWpPG2ORM/QRNyW42aLxcak=
-github.com/atterpac/dado v0.2.9/go.mod h1:oBqIXWzQMu1DGh8sAuyBAM5wHrIz1anqPoa9jcfnTJE=
+github.com/atterpac/dado v0.2.10 h1:RELhKjHEHOz2SFULSeU2YQGqSZY7nCbfuLwWSIz2ies=
+github.com/atterpac/dado v0.2.10/go.mod h1:oBqIXWzQMu1DGh8sAuyBAM5wHrIz1anqPoa9jcfnTJE=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

internal/nats/client.go

@@ -877,7 +877,19 @@ func (c *Client) SubscribeJetStream(ctx context.Context, subject string, policy
 	if err != nil {
 		return nil, fmt.Errorf("no stream found for subject %q: %w", subject, err)
 	}
+	return c.subscribeJetStream(ctx, streamName, subject, policy, handler)
+}
+
+// SubscribeJetStreamStream creates a JetStream subscription over an entire
+// stream. The ephemeral consumer carries no subject filter, so every message
+// in the stream is delivered regardless of which configured subject it matched.
+func (c *Client) SubscribeJetStreamStream(ctx context.Context, streamName string, policy DeliverPolicy, handler func(LiveMessage)) (Subscription, error) {
+	return c.subscribeJetStream(ctx, streamName, "", policy, handler)
+}
 
+// subscribeJetStream creates an ephemeral consumer on streamName. An empty
+// filterSubject watches the whole stream; a non-empty one narrows to that subject.
+func (c *Client) subscribeJetStream(ctx context.Context, streamName, filterSubject string, policy DeliverPolicy, handler func(LiveMessage)) (Subscription, error) {
 	stream, err := c.jetStream().Stream(ctx, streamName)
 	if err != nil {
 		return nil, fmt.Errorf("getting stream %s: %w", streamName, err)
@@ -898,9 +910,9 @@ func (c *Client) SubscribeJetStream(ctx context.Context, subject string, policy
 		jsPolicy = jetstream.DeliverAllPolicy
 	}
 
-	// Create ephemeral consumer
+	// Create ephemeral consumer. An empty FilterSubject delivers all subjects.
 	consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
-		FilterSubject:     subject,
+		FilterSubject:     filterSubject,
 		DeliverPolicy:     jsPolicy,
 		AckPolicy:         jetstream.AckNonePolicy, // No acks needed for monitoring
 		InactiveThreshold: 5 * time.Minute,

internal/nats/provider.go

@@ -108,6 +108,9 @@ type Provider interface {
 
 	// JetStream subscription with replay capability
 	SubscribeJetStream(ctx context.Context, subject string, policy DeliverPolicy, handler func(LiveMessage)) (Subscription, error)
+
+	// JetStream subscription over an entire stream (no subject filter — all subjects)
+	SubscribeJetStreamStream(ctx context.Context, streamName string, policy DeliverPolicy, handler func(LiveMessage)) (Subscription, error)
 }
 
 // ConnectionStats holds basic connection statistics.

internal/view/app.go

@@ -157,6 +157,7 @@ func (a *App) setup() {
 			}
 			if a.app.Pages().CanPop() {
 				a.app.Pages().Pop()
+				a.app.SetFocus(a.app.Pages())
 				return nil
 			}
 			return event
@@ -310,6 +311,11 @@ func (a *App) NavigateToMessageMonitorWithSubject(subject string) {
 	a.app.Pages().Push(view)
 }
 
+func (a *App) NavigateToMessageMonitorForStream(streamName string) {
+	view := NewMessageMonitorForStream(a, streamName)
+	a.app.Pages().Push(view)
+}
+
 func (a *App) NavigateToConsumerLag() {
 	view := NewConsumerLag(a)
 	a.app.Pages().Push(view)

internal/view/message_monitor.go

@@ -79,6 +79,10 @@ type MessageMonitor struct {
 	// Initial subject (for auto-subscribe on Start)
 	initSubject string
 
+	// When set, the monitor watches this entire stream (all subjects) instead
+	// of a single subject filter. The subject input becomes a client-side filter.
+	streamName string
+
 	// JSON path / pipeline filter for preview pane
 	jsonFilter string
 	pipeline   *Pipeline
@@ -89,6 +93,24 @@ func NewMessageMonitor(app *App) *MessageMonitor {
 	return NewMessageMonitorWithSubject(app, "")
 }
 
+// NewMessageMonitorForStream creates a monitor that watches an entire stream
+// (every subject it stores), rather than a single subject filter. The subject
+// field acts as a client-side filter over the live messages instead of
+// changing the subscription.
+func NewMessageMonitorForStream(app *App, streamName string) *MessageMonitor {
+	mm := &MessageMonitor{
+		app:           app,
+		useJetStream:  true,
+		deliverPolicy: nats.DeliverAll,
+		streamName:    streamName,
+		processCancel: func() {},
+	}
+
+	mm.buildUI("")
+	mm.setupBinding()
+	return mm
+}
+
 // NewMessageMonitorWithSubject creates the message monitor with a pre-filled subject.
 func NewMessageMonitorWithSubject(app *App, subject string) *MessageMonitor {
 	mm := &MessageMonitor{
@@ -123,16 +145,30 @@ func (mm *MessageMonitor) buildUI(subject string) {
 		SetWordWrap(true)
 	mm.preview.SetBackgroundColor(theme.Bg())
 
-	// Subject input
+	// Subject input. In stream-watch mode it filters the live list client-side
+	// rather than re-subscribing, since the subscription already covers the
+	// whole stream.
+	subjectLabel := "Subject: "
+	subjectPlaceholder := ">"
+	if mm.streamName != "" {
+		subjectLabel = "Filter: "
+		subjectPlaceholder = "filter subjects..."
+	}
 	mm.subjectInput = components.NewTextField("subject").
-		SetLabel("Subject: ").
-		SetPlaceholder(">")
+		SetLabel(subjectLabel).
+		SetPlaceholder(subjectPlaceholder)
 	mm.subjectInput.SetBackgroundColor(theme.Bg())
 	mm.subjectInput.SetOnSubmit(func(_ *components.SubmitEvent) {
-		subj := mm.subjectInput.GetValue()
-		// subscribe is goroutine-safe; dispatch off main goroutine so
-		// the NATS handshake cannot block the event loop.
-		go mm.subscribe(subj)
+		val := mm.subjectInput.GetValue()
+		// Move focus off the text input so arrow keys navigate the message
+		// list instead of being captured by the field.
+		mm.app.app.SetFocus(mm.MasterDetailView)
+		// Dispatch off the main goroutine so the work can't block the event loop.
+		if mm.streamName != "" {
+			go mm.setFilter(val)
+			return
+		}
+		go mm.subscribe(val)
 	})
 	if subject != "" {
 		mm.subjectInput.SetValue(subject)
@@ -237,6 +273,12 @@ func (mm *MessageMonitor) Start() {
 	// Start() runs on main goroutine — safe to touch UI directly.
 	atomic.StoreInt32(&mm.stopped, 0)
 
+	// Whole-stream watch: subscribe to the entire stream regardless of subject.
+	if mm.streamName != "" {
+		go mm.subscribe("")
+		return
+	}
+
 	// Re-subscribe to whatever subject was active (or the initial one).
 	subject := mm.state.Get().subject
 	if subject == "" {
@@ -434,10 +476,23 @@ func (mm *MessageMonitor) HandleKey(event *tcell.EventKey) bool {
 // ── Subscribe / Unsubscribe ────────────────────────────────────────────────
 
 func (mm *MessageMonitor) subscribe(subject string) {
-	if subject == "" || atomic.LoadInt32(&mm.stopped) == 1 {
+	if atomic.LoadInt32(&mm.stopped) == 1 {
 		return
 	}
 
+	// Whole-stream watch mode: no subject filter, drive off the stream name.
+	streamMode := mm.streamName != "" && subject == ""
+	if subject == "" && !streamMode {
+		return
+	}
+
+	// Label used for display/state (the subscription itself is by stream in
+	// stream mode, so there is no single subject to show).
+	displayLabel := subject
+	if streamMode {
+		displayLabel = fmt.Sprintf("%s (all subjects)", mm.streamName)
+	}
+
 	// Tear down any existing subscription first.
 	mm.doUnsubscribe()
 
@@ -461,15 +516,15 @@ func (mm *MessageMonitor) subscribe(subject string) {
 	// Push "subscribing" state.
 	mm.state.SetAndDraw(monitorState{
 		status:  statusSubscribing,
-		subject: subject,
+		subject: displayLabel,
 		mode:    mm.modeLabel(),
 		policy:  mm.policyLabel(),
 	})
 
 	// Start processor goroutine for this subscription.
 	processCtx, processCancel := context.WithCancel(context.Background())
 	mm.processCancel = processCancel
-	go mm.processMessages(processCtx, myGen, msgChan, subject)
+	go mm.processMessages(processCtx, myGen, msgChan, displayLabel)
 
 	// Handler: never blocks, never holds locks. Channel send is guarded by
 	// recover so a closed-channel panic in the narrow unsubscribe race is safe.
@@ -489,16 +544,21 @@ func (mm *MessageMonitor) subscribe(subject string) {
 	var err error
 	var modeLabel string
 
-	if useJS {
+	if streamMode {
+		// Whole-stream watch is inherently JetStream — no single subject for
+		// a core NATS subscription to bind to.
+		sub, err = provider.SubscribeJetStreamStream(context.Background(), mm.streamName, policy, handler)
+		modeLabel = "JetStream"
+	} else if useJS {
 		sub, err = provider.SubscribeJetStream(context.Background(), subject, policy, handler)
 		modeLabel = "JetStream"
 	} else {
 		sub, err = provider.Subscribe(context.Background(), subject, handler)
 		modeLabel = "Core NATS"
 	}
 
-	// Fallback to core NATS if JetStream fails.
-	if err != nil && useJS {
+	// Fallback to core NATS if JetStream fails (single-subject mode only).
+	if err != nil && useJS && !streamMode {
 		sub, err = provider.Subscribe(context.Background(), subject, handler)
 		if err == nil {
 			modeLabel = "Core NATS (fallback)"
@@ -511,7 +571,7 @@ func (mm *MessageMonitor) subscribe(subject string) {
 	if err != nil {
 		mm.state.SetAndDraw(monitorState{
 			status:  statusError,
-			subject: subject,
+			subject: displayLabel,
 			errMsg:  err.Error(),
 			mode:    mm.modeLabel(),
 			policy:  mm.policyLabel(),
@@ -533,7 +593,7 @@ func (mm *MessageMonitor) subscribe(subject string) {
 	mm.mu.Unlock()
 
 	mm.app.QueueUpdateDraw(func() {
-		mm.app.ShowSuccess(fmt.Sprintf("Subscribed to %s (%s)", subject, modeLabel))
+		mm.app.ShowSuccess(fmt.Sprintf("Subscribed to %s (%s)", displayLabel, modeLabel))
 	})
 }
 

internal/view/stream_detail.go

@@ -137,8 +137,8 @@ func (sd *StreamDetail) HandleKey(event *tcell.EventKey) bool {
 		go sd.showSubjectBreakdown()
 		return true
 	case 'w':
-		if info := sd.info.Get(); info != nil && len(info.Config.Subjects) > 0 {
-			sd.app.NavigateToMessageMonitorWithSubject(info.Config.Subjects[0])
+		if info := sd.info.Get(); info != nil {
+			sd.app.NavigateToMessageMonitorForStream(info.Config.Name)
 		}
 		return true
 	case 'e':

internal/view/stream_list.go

@@ -85,10 +85,8 @@ func NewStreamList(app *App) *StreamList {
 		}).
 		SetRefreshInterval(10 * time.Second).
 		SetOnSelect(func(s *jetstream.StreamInfo) {
-			// Enter opens watch view directly
-			if len(s.Config.Subjects) > 0 {
-				sl.app.NavigateToMessageMonitorWithSubject(s.Config.Subjects[0])
-			}
+			// Enter opens watch view directly — watch the whole stream.
+			sl.app.NavigateToMessageMonitorForStream(s.Config.Name)
 		}).
 		SetOnRefresh(func(data []*jetstream.StreamInfo, err error) {
 			if err != nil {
@@ -201,8 +199,8 @@ func (sl *StreamList) Hints() []components.KeyHint {
 func (sl *StreamList) HandleKey(event *tcell.EventKey) bool {
 	switch {
 	case event.Key() == tcell.KeyEnter:
-		if s, ok := sl.binding.GetSelectedValue(); ok && s != nil && len(s.Config.Subjects) > 0 {
-			sl.app.NavigateToMessageMonitorWithSubject(s.Config.Subjects[0])
+		if s, ok := sl.binding.GetSelectedValue(); ok && s != nil {
+			sl.app.NavigateToMessageMonitorForStream(s.Config.Name)
 		}
 		return true
 	case event.Rune() == 'v':