feat: add generic pub/sub helpers and pubsub-crud example

Add typed Publish[T] and Subscribe[T] generic helpers that handle
JSON marshaling, along with vianats.EnsureStream and ReplayHistory
helpers. Refactor nats-chatroom to use the new APIs.

Add pubsub-crud example demonstrating CRUD operations with DaisyUI
toast notifications broadcast to all connected clients via NATS.

actiontrigger.go

@@ -21,6 +21,8 @@ type triggerOpts struct {
 	hasSignal      bool
 	signalID       string
 	value          string
+	window         bool
+	preventDefault bool
 }
 
 type withSignalOpt struct {
@@ -34,6 +36,28 @@ func (o withSignalOpt) apply(opts *triggerOpts) {
 	opts.value = o.value
 }
 
+type withWindowOpt struct{}
+
+func (o withWindowOpt) apply(opts *triggerOpts) {
+	opts.window = true
+}
+
+// WithWindow makes the event listener attach to the window instead of the element.
+func WithWindow() ActionTriggerOption {
+	return withWindowOpt{}
+}
+
+type withPreventDefaultOpt struct{}
+
+func (o withPreventDefaultOpt) apply(opts *triggerOpts) {
+	opts.preventDefault = true
+}
+
+// WithPreventDefault calls evt.preventDefault() for matched keys.
+func WithPreventDefault() ActionTriggerOption {
+	return withPreventDefaultOpt{}
+}
+
 // WithSignal sets a signal value before triggering the action.
 func WithSignal(sig *signal, value string) ActionTriggerOption {
 	return withSignalOpt{
@@ -54,7 +78,7 @@ func buildOnExpr(base string, opts *triggerOpts) string {
 	if !opts.hasSignal {
 		return base
 	}
-	return fmt.Sprintf("$%s=%s;%s", opts.signalID, opts.value, base)
+	return fmt.Sprintf("$%s=%s,%s", opts.signalID, opts.value, base)
 }
 
 func applyOptions(options ...ActionTriggerOption) triggerOpts {
@@ -92,5 +116,49 @@ func (a *actionTrigger) OnKeyDown(key string, options ...ActionTriggerOption) h.
 	if key != "" {
 		condition = fmt.Sprintf("evt.key==='%s' &&", key)
 	}
-	return h.Data("on:keydown", fmt.Sprintf("%s%s", condition, buildOnExpr(actionURL(a.id), &opts)))
+	attrName := "on:keydown"
+	if opts.window {
+		attrName = "on:keydown__window"
+	}
+	return h.Data(attrName, fmt.Sprintf("%s%s", condition, buildOnExpr(actionURL(a.id), &opts)))
+}
+
+// KeyBinding pairs a key with an action and per-binding options.
+type KeyBinding struct {
+	Key     string
+	Action  *actionTrigger
+	Options []ActionTriggerOption
+}
+
+// KeyBind creates a KeyBinding for use with OnKeyDownMap.
+func KeyBind(key string, action *actionTrigger, options ...ActionTriggerOption) KeyBinding {
+	return KeyBinding{Key: key, Action: action, Options: options}
+}
+
+// OnKeyDownMap produces a single window-scoped keydown attribute that dispatches
+// to different actions based on the pressed key. Each binding can reference a
+// different action and carry its own signal/preventDefault options.
+func OnKeyDownMap(bindings ...KeyBinding) h.H {
+	if len(bindings) == 0 {
+		return nil
+	}
+
+	expr := ""
+	for i, b := range bindings {
+		opts := applyOptions(b.Options...)
+
+		branch := ""
+		if opts.preventDefault {
+			branch = "evt.preventDefault(),"
+		}
+		branch += buildOnExpr(actionURL(b.Action.id), &opts)
+
+		if i > 0 {
+			expr += " : "
+		}
+		expr += fmt.Sprintf("evt.key==='%s' ? (%s)", b.Key, branch)
+	}
+	expr += " : void 0"
+
+	return h.Data("on:keydown__window", expr)
 }

internal/examples/keyboard/main.go

@@ -0,0 +1,74 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/ryanhamamura/via"
+	"github.com/ryanhamamura/via/h"
+)
+
+const gridSize = 8
+
+func main() {
+	v := via.New()
+	v.Config(via.Options{DocumentTitle: "Keyboard", ServerAddress: ":7331"})
+
+	v.Page("/", func(c *via.Context) {
+		x, y := 0, 0
+		dir := c.Signal("")
+
+		move := c.Action(func() {
+			switch dir.String() {
+			case "up":
+				y = max(0, y-1)
+			case "down":
+				y = min(gridSize-1, y+1)
+			case "left":
+				x = max(0, x-1)
+			case "right":
+				x = min(gridSize-1, x+1)
+			}
+			c.Sync()
+		})
+
+		c.View(func() h.H {
+			var rows []h.H
+			for row := range gridSize {
+				var cells []h.H
+				for col := range gridSize {
+					bg := "#e0e0e0"
+					if col == x && row == y {
+						bg = "#4a90d9"
+					}
+					cells = append(cells, h.Div(
+						h.Attr("style", fmt.Sprintf(
+							"width:48px;height:48px;background:%s;border:1px solid #ccc;",
+							bg,
+						)),
+					))
+				}
+				rows = append(rows, h.Div(
+					append([]h.H{h.Attr("style", "display:flex;")}, cells...)...,
+				))
+			}
+
+			return h.Div(
+				h.H1(h.Text("Keyboard Grid")),
+				h.P(h.Text("Move with WASD or arrow keys")),
+				h.Div(rows...),
+				via.OnKeyDownMap(
+					via.KeyBind("w", move, via.WithSignal(dir, "up")),
+					via.KeyBind("a", move, via.WithSignal(dir, "left")),
+					via.KeyBind("s", move, via.WithSignal(dir, "down")),
+					via.KeyBind("d", move, via.WithSignal(dir, "right")),
+					via.KeyBind("ArrowUp", move, via.WithSignal(dir, "up"), via.WithPreventDefault()),
+					via.KeyBind("ArrowLeft", move, via.WithSignal(dir, "left"), via.WithPreventDefault()),
+					via.KeyBind("ArrowDown", move, via.WithSignal(dir, "down"), via.WithPreventDefault()),
+					via.KeyBind("ArrowRight", move, via.WithSignal(dir, "right"), via.WithPreventDefault()),
+				),
+			)
+		})
+	})
+
+	v.Start()
+}

internal/examples/nats-chatroom/main.go

@@ -2,13 +2,11 @@ package main
 
 import (
 	"context"
-	"encoding/json"
 	"log"
 	"math/rand"
 	"sync"
 	"time"
 
-	"github.com/nats-io/nats.go"
 	"github.com/ryanhamamura/via"
 	"github.com/ryanhamamura/via/h"
 	"github.com/ryanhamamura/via/vianats"
@@ -46,15 +44,15 @@ func main() {
 	}
 	defer ps.Close()
 
-	// Create JetStream stream for message durability
-	js := ps.JetStream()
-	js.AddStream(&nats.StreamConfig{
+	err = vianats.EnsureStream(ps, vianats.StreamConfig{
 		Name:     "CHAT",
 		Subjects: []string{"chat.>"},
-		Retention: nats.LimitsPolicy,
 		MaxMsgs:  1000,
 		MaxAge:   24 * time.Hour,
 	})
+	if err != nil {
+		log.Fatalf("Failed to ensure stream: %v", err)
+	}
 
 	v := via.New()
 	v.Config(via.Options{
@@ -147,30 +145,14 @@ func main() {
 				currentSub.Unsubscribe()
 			}
 
-			// Replay history from JetStream before subscribing for real-time
 			subject := "chat.room." + room
-			if hist, err := js.SubscribeSync(subject, nats.DeliverAll(), nats.OrderedConsumer()); err == nil {
-				for {
-					msg, err := hist.NextMsg(200 * time.Millisecond)
-					if err != nil {
-						break
-					}
-					var chatMsg ChatMessage
-					if json.Unmarshal(msg.Data, &chatMsg) == nil {
-						messages = append(messages, chatMsg)
-					}
-				}
-				hist.Unsubscribe()
-				if len(messages) > 50 {
-					messages = messages[len(messages)-50:]
-				}
+
+			// Replay history from JetStream
+			if hist, err := vianats.ReplayHistory[ChatMessage](ps, subject, 50); err == nil {
+				messages = hist
 			}
 
-			sub, _ := c.Subscribe(subject, func(data []byte) {
-				var msg ChatMessage
-				if err := json.Unmarshal(data, &msg); err != nil {
-					return
-				}
+			sub, _ := via.Subscribe(c, subject, func(msg ChatMessage) {
 				messagesMu.Lock()
 				messages = append(messages, msg)
 				if len(messages) > 50 {
@@ -203,12 +185,11 @@ func main() {
 			}
 			statement.SetValue("")
 
-			data, _ := json.Marshal(ChatMessage{
+			via.Publish(c, "chat.room."+currentRoom, ChatMessage{
 				User:    currentUser,
 				Message: msg,
 				Time:    time.Now().UnixMilli(),
 			})
-			c.Publish("chat.room."+currentRoom, data)
 		})
 
 		c.View(func() h.H {

internal/examples/pubsub-crud/main.go

@@ -0,0 +1,284 @@
+package main
+
+import (
+	"context"
+	"crypto/rand"
+	"fmt"
+	"html"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/ryanhamamura/via"
+	"github.com/ryanhamamura/via/h"
+	"github.com/ryanhamamura/via/vianats"
+)
+
+var WithSignal = via.WithSignal
+
+type Bookmark struct {
+	ID    string
+	Title string
+	URL   string
+}
+
+type CRUDEvent struct {
+	Action string `json:"action"`
+	Title  string `json:"title"`
+	UserID string `json:"user_id"`
+}
+
+var (
+	bookmarks   []Bookmark
+	bookmarksMu sync.RWMutex
+)
+
+func randomHex(n int) string {
+	b := make([]byte, n)
+	rand.Read(b)
+	return fmt.Sprintf("%x", b)
+}
+
+func findBookmark(id string) (Bookmark, int) {
+	for i, bm := range bookmarks {
+		if bm.ID == id {
+			return bm, i
+		}
+	}
+	return Bookmark{}, -1
+}
+
+func main() {
+	ctx := context.Background()
+
+	ps, err := vianats.New(ctx, "./data/nats")
+	if err != nil {
+		log.Fatalf("Failed to start embedded NATS: %v", err)
+	}
+	defer ps.Close()
+
+	err = vianats.EnsureStream(ps, vianats.StreamConfig{
+		Name:     "BOOKMARKS",
+		Subjects: []string{"bookmarks.>"},
+		MaxMsgs:  1000,
+		MaxAge:   24 * time.Hour,
+	})
+	if err != nil {
+		log.Fatalf("Failed to ensure stream: %v", err)
+	}
+
+	v := via.New()
+	v.Config(via.Options{
+		DevMode:       true,
+		DocumentTitle: "Bookmarks",
+		LogLevel:      via.LogLevelInfo,
+		ServerAddress: ":7331",
+		PubSub:        ps,
+	})
+
+	v.AppendToHead(
+		h.Link(h.Rel("stylesheet"), h.Href("https://cdn.jsdelivr.net/npm/daisyui@4/dist/full.min.css")),
+		h.Script(h.Src("https://cdn.tailwindcss.com")),
+	)
+
+	v.Page("/", func(c *via.Context) {
+		userID := randomHex(8)
+
+		titleSignal := c.Signal("")
+		urlSignal := c.Signal("")
+		targetIDSignal := c.Signal("")
+
+		via.Subscribe(c, "bookmarks.events", func(evt CRUDEvent) {
+			if evt.UserID == userID {
+				return
+			}
+			safeTitle := html.EscapeString(evt.Title)
+			var alertClass string
+			switch evt.Action {
+			case "created":
+				alertClass = "alert-success"
+			case "updated":
+				alertClass = "alert-info"
+			case "deleted":
+				alertClass = "alert-error"
+			}
+			c.ExecScript(fmt.Sprintf(`(function(){
+				var tc = document.getElementById('toast-container');
+				if (!tc) return;
+				var d = document.createElement('div');
+				d.className = 'alert %s';
+				d.innerHTML = '<span>Bookmark "%s" %s</span>';
+				tc.appendChild(d);
+				setTimeout(function(){ d.remove(); }, 3000);
+			})()`, alertClass, safeTitle, evt.Action))
+			c.Sync()
+		})
+
+		save := c.Action(func() {
+			title := titleSignal.String()
+			url := urlSignal.String()
+			if title == "" || url == "" {
+				return
+			}
+
+			targetID := targetIDSignal.String()
+			action := "created"
+
+			bookmarksMu.Lock()
+			if targetID != "" {
+				if _, idx := findBookmark(targetID); idx >= 0 {
+					bookmarks[idx].Title = title
+					bookmarks[idx].URL = url
+					action = "updated"
+				}
+			} else {
+				bookmarks = append(bookmarks, Bookmark{
+					ID:    randomHex(8),
+					Title: title,
+					URL:   url,
+				})
+			}
+			bookmarksMu.Unlock()
+
+			titleSignal.SetValue("")
+			urlSignal.SetValue("")
+			targetIDSignal.SetValue("")
+
+			via.Publish(c, "bookmarks.events", CRUDEvent{
+				Action: action,
+				Title:  title,
+				UserID: userID,
+			})
+			c.Sync()
+		})
+
+		edit := c.Action(func() {
+			id := targetIDSignal.String()
+			bookmarksMu.RLock()
+			bm, idx := findBookmark(id)
+			bookmarksMu.RUnlock()
+			if idx < 0 {
+				return
+			}
+			titleSignal.SetValue(bm.Title)
+			urlSignal.SetValue(bm.URL)
+		})
+
+		del := c.Action(func() {
+			id := targetIDSignal.String()
+			bookmarksMu.Lock()
+			bm, idx := findBookmark(id)
+			if idx >= 0 {
+				bookmarks = append(bookmarks[:idx], bookmarks[idx+1:]...)
+			}
+			bookmarksMu.Unlock()
+			if idx < 0 {
+				return
+			}
+
+			targetIDSignal.SetValue("")
+
+			via.Publish(c, "bookmarks.events", CRUDEvent{
+				Action: "deleted",
+				Title:  bm.Title,
+				UserID: userID,
+			})
+			c.Sync()
+		})
+
+		cancelEdit := c.Action(func() {
+			titleSignal.SetValue("")
+			urlSignal.SetValue("")
+			targetIDSignal.SetValue("")
+		})
+
+		c.View(func() h.H {
+			isEditing := targetIDSignal.String() != ""
+
+			// Build table rows
+			bookmarksMu.RLock()
+			var rows []h.H
+			for _, bm := range bookmarks {
+				rows = append(rows, h.Tr(
+					h.Td(h.Text(bm.Title)),
+					h.Td(h.A(h.Href(bm.URL), h.Attr("target", "_blank"), h.Class("link link-primary"), h.Text(bm.URL))),
+					h.Td(
+						h.Div(h.Class("flex gap-1"),
+							h.Button(h.Class("btn btn-xs btn-ghost"), h.Text("Edit"),
+								edit.OnClick(WithSignal(targetIDSignal, bm.ID)),
+							),
+							h.Button(h.Class("btn btn-xs btn-ghost text-error"), h.Text("Delete"),
+								del.OnClick(WithSignal(targetIDSignal, bm.ID)),
+							),
+						),
+					),
+				))
+			}
+			bookmarksMu.RUnlock()
+
+			saveLabel := "Add Bookmark"
+			if isEditing {
+				saveLabel = "Update Bookmark"
+			}
+
+			return h.Div(h.Class("min-h-screen bg-base-200"),
+				// Navbar
+				h.Div(h.Class("navbar bg-base-100 shadow-sm"),
+					h.Div(h.Class("flex-1"),
+						h.A(h.Class("btn btn-ghost text-xl"), h.Text("Bookmarks")),
+					),
+					h.Div(h.Class("flex-none"),
+						h.Div(h.Class("badge badge-outline"), h.Text(userID[:8])),
+					),
+				),
+
+				h.Div(h.Class("container mx-auto p-4 max-w-3xl flex flex-col gap-4"),
+					// Form card
+					h.Div(h.Class("card bg-base-100 shadow"),
+						h.Div(h.Class("card-body"),
+							h.H2(h.Class("card-title"), h.Text(saveLabel)),
+							h.Div(h.Class("flex flex-col gap-2"),
+								h.Input(h.Class("input input-bordered w-full"), h.Type("text"), h.Placeholder("Title"), titleSignal.Bind()),
+								h.Input(h.Class("input input-bordered w-full"), h.Type("text"), h.Placeholder("https://example.com"), urlSignal.Bind()),
+								h.Div(h.Class("card-actions justify-end"),
+									h.If(isEditing,
+										h.Button(h.Class("btn btn-ghost"), h.Text("Cancel"), cancelEdit.OnClick()),
+									),
+									h.Button(h.Class("btn btn-primary"), h.Text(saveLabel), save.OnClick()),
+								),
+							),
+						),
+					),
+
+					// Table card
+					h.Div(h.Class("card bg-base-100 shadow"),
+						h.Div(h.Class("card-body"),
+							h.H2(h.Class("card-title"), h.Text("All Bookmarks")),
+							h.If(len(rows) == 0,
+								h.P(h.Class("text-base-content/60"), h.Text("No bookmarks yet. Add one above!")),
+							),
+							h.If(len(rows) > 0,
+								h.Div(h.Class("overflow-x-auto"),
+									h.Table(h.Class("table"),
+										h.THead(h.Tr(
+											h.Th(h.Text("Title")),
+											h.Th(h.Text("URL")),
+											h.Th(h.Text("Actions")),
+										)),
+										h.TBody(rows...),
+									),
+								),
+							),
+						),
+					),
+				),
+
+				// Toast container — ignored by morph so Sync() doesn't wipe active toasts
+				h.Div(h.ID("toast-container"), h.Class("toast toast-end toast-top"), h.DataIgnoreMorph()),
+			)
+		})
+	})
+
+	log.Println("Starting pubsub-crud example on :7331")
+	v.Start()
+}

pubsub_helpers.go

@@ -0,0 +1,23 @@
+package via
+
+import "encoding/json"
+
+// Publish JSON-marshals msg and publishes to subject.
+func Publish[T any](c *Context, subject string, msg T) error {
+	data, err := json.Marshal(msg)
+	if err != nil {
+		return err
+	}
+	return c.Publish(subject, data)
+}
+
+// Subscribe JSON-unmarshals each message as T and calls handler.
+func Subscribe[T any](c *Context, subject string, handler func(T)) (Subscription, error) {
+	return c.Subscribe(subject, func(data []byte) {
+		var msg T
+		if err := json.Unmarshal(data, &msg); err != nil {
+			return
+		}
+		handler(msg)
+	})
+}

pubsub_helpers_test.go

@@ -0,0 +1,66 @@
+package via
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/ryanhamamura/via/h"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPublishSubscribe_RoundTrip(t *testing.T) {
+	ps := newMockPubSub()
+	v := New()
+	v.Config(Options{PubSub: ps})
+
+	type event struct {
+		Name  string `json:"name"`
+		Count int    `json:"count"`
+	}
+
+	var got event
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	c := newContext("typed-ctx", "/", v)
+	c.View(func() h.H { return h.Div() })
+
+	_, err := Subscribe(c, "events", func(e event) {
+		got = e
+		wg.Done()
+	})
+	require.NoError(t, err)
+
+	err = Publish(c, "events", event{Name: "click", Count: 42})
+	require.NoError(t, err)
+
+	wg.Wait()
+	assert.Equal(t, "click", got.Name)
+	assert.Equal(t, 42, got.Count)
+}
+
+func TestSubscribe_SkipsBadJSON(t *testing.T) {
+	ps := newMockPubSub()
+	v := New()
+	v.Config(Options{PubSub: ps})
+
+	type msg struct {
+		Text string `json:"text"`
+	}
+
+	called := false
+	c := newContext("bad-json-ctx", "/", v)
+	c.View(func() h.H { return h.Div() })
+
+	_, err := Subscribe(c, "topic", func(m msg) {
+		called = true
+	})
+	require.NoError(t, err)
+
+	// Publish raw invalid JSON — handler should silently skip
+	err = c.Publish("topic", []byte("not json"))
+	require.NoError(t, err)
+
+	assert.False(t, called)
+}

via_test.go

@@ -128,6 +128,101 @@ func TestAction(t *testing.T) {
 	assert.Contains(t, body, "/_action/")
 }
 
+func TestOnKeyDownWithWindow(t *testing.T) {
+	var trigger *actionTrigger
+	v := New()
+	v.Page("/", func(c *Context) {
+		trigger = c.Action(func() {})
+		c.View(func() h.H {
+			return h.Div(trigger.OnKeyDown("Enter", WithWindow()))
+		})
+	})
+
+	req := httptest.NewRequest("GET", "/", nil)
+	w := httptest.NewRecorder()
+	v.mux.ServeHTTP(w, req)
+	body := w.Body.String()
+	assert.Contains(t, body, "data-on:keydown__window")
+	assert.Contains(t, body, "evt.key==='Enter'")
+}
+
+func TestOnKeyDownMap(t *testing.T) {
+	t.Run("multiple bindings with different actions", func(t *testing.T) {
+		var move, shoot *actionTrigger
+		var dir *signal
+		v := New()
+		v.Page("/", func(c *Context) {
+			dir = c.Signal("none")
+			move = c.Action(func() {})
+			shoot = c.Action(func() {})
+			c.View(func() h.H {
+				return h.Div(
+					OnKeyDownMap(
+						KeyBind("w", move, WithSignal(dir, "up")),
+						KeyBind("ArrowUp", move, WithSignal(dir, "up"), WithPreventDefault()),
+						KeyBind(" ", shoot, WithPreventDefault()),
+					),
+				)
+			})
+		})
+
+		req := httptest.NewRequest("GET", "/", nil)
+		w := httptest.NewRecorder()
+		v.mux.ServeHTTP(w, req)
+		body := w.Body.String()
+
+		// Single attribute, window-scoped
+		assert.Contains(t, body, "data-on:keydown__window")
+
+		// Key dispatching
+		assert.Contains(t, body, "evt.key==='w'")
+		assert.Contains(t, body, "evt.key==='ArrowUp'")
+		assert.Contains(t, body, "evt.key===' '")
+
+		// Different actions referenced
+		assert.Contains(t, body, "/_action/"+move.id)
+		assert.Contains(t, body, "/_action/"+shoot.id)
+
+		// preventDefault only on ArrowUp and space branches
+		assert.Contains(t, body, "evt.key==='ArrowUp' ? (evt.preventDefault()")
+		assert.Contains(t, body, "evt.key===' ' ? (evt.preventDefault()")
+
+		// 'w' branch should NOT have preventDefault
+		assert.NotContains(t, body, "evt.key==='w' ? (evt.preventDefault()")
+	})
+
+	t.Run("WithSignal per binding", func(t *testing.T) {
+		var move *actionTrigger
+		var dir *signal
+		v := New()
+		v.Page("/", func(c *Context) {
+			dir = c.Signal("none")
+			move = c.Action(func() {})
+			c.View(func() h.H {
+				return h.Div(
+					OnKeyDownMap(
+						KeyBind("w", move, WithSignal(dir, "up")),
+						KeyBind("s", move, WithSignal(dir, "down")),
+					),
+				)
+			})
+		})
+
+		req := httptest.NewRequest("GET", "/", nil)
+		w := httptest.NewRecorder()
+		v.mux.ServeHTTP(w, req)
+		body := w.Body.String()
+
+		assert.Contains(t, body, "$"+dir.ID()+"='up'")
+		assert.Contains(t, body, "$"+dir.ID()+"='down'")
+	})
+
+	t.Run("empty bindings returns nil", func(t *testing.T) {
+		result := OnKeyDownMap()
+		assert.Nil(t, result)
+	})
+}
+
 func TestConfig(t *testing.T) {
 	v := New()
 	v.Config(Options{DocumentTitle: "Test"})

vianats/vianats.go

@@ -4,7 +4,9 @@ package vianats
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"time"
 
 	"github.com/delaneyj/toolbelt/embeddednats"
 	"github.com/nats-io/nats.go"
@@ -76,3 +78,50 @@ func (n *NATS) Conn() *nats.Conn {
 func (n *NATS) JetStream() nats.JetStreamContext {
 	return n.js
 }
+
+// StreamConfig holds the parameters for creating or updating a JetStream stream.
+type StreamConfig struct {
+	Name     string
+	Subjects []string
+	MaxMsgs  int64
+	MaxAge   time.Duration
+}
+
+// EnsureStream creates or updates a JetStream stream matching cfg.
+func EnsureStream(n *NATS, cfg StreamConfig) error {
+	_, err := n.js.AddStream(&nats.StreamConfig{
+		Name:      cfg.Name,
+		Subjects:  cfg.Subjects,
+		Retention: nats.LimitsPolicy,
+		MaxMsgs:   cfg.MaxMsgs,
+		MaxAge:    cfg.MaxAge,
+	})
+	return err
+}
+
+// ReplayHistory fetches the last limit messages from subject,
+// deserializing each as T. Returns an empty slice if nothing is available.
+func ReplayHistory[T any](n *NATS, subject string, limit int) ([]T, error) {
+	sub, err := n.js.SubscribeSync(subject, nats.DeliverAll(), nats.OrderedConsumer())
+	if err != nil {
+		return nil, err
+	}
+	defer sub.Unsubscribe()
+
+	var msgs []T
+	for {
+		raw, err := sub.NextMsg(200 * time.Millisecond)
+		if err != nil {
+			break
+		}
+		var msg T
+		if json.Unmarshal(raw.Data, &msg) == nil {
+			msgs = append(msgs, msg)
+		}
+	}
+
+	if limit > 0 && len(msgs) > limit {
+		msgs = msgs[len(msgs)-limit:]
+	}
+	return msgs, nil
+}