feat: add declarative Options.Streams for automatic JetStream stream creation

Streams listed in Options.Streams are created by Start() when the
embedded NATS server initializes, replacing manual EnsureStream calls
during setup. Migrates nats-chatroom and pubsub-crud examples.

.claude/commands/release.md

@@ -0,0 +1,14 @@
+Create a new release for this project. Steps:
+
+1. Fetch tags from all remotes so the version list is current.
+2. Check for uncommitted changes. If any exist, commit them with a clean semantic commit message. No Claude attribution lines.
+3. Review the commits since the last tag. Based on their content, recommend a semver bump:
+   - **major**: breaking/incompatible API changes
+   - **minor**: new features, meaningful new behavior
+   - **patch**: bug fixes, docs, refactoring with no new features
+   Present the proposed version, the bump rationale, and the commit list. Wait for user approval before continuing.
+4. Tag the new version and push the tag + commits to all remotes (origin, gitea, etc.).
+5. Generate release notes from the commits since the last tag, grouped by type (features, fixes, docs/refactoring).
+6. Create a GitHub release using `gh release create`.
+7. Create a Gitea release using `tea releases create` with the same notes.
+8. Report both release URLs and confirm all remotes are up to date.

actiontrigger.go

@@ -3,6 +3,7 @@ package via
 import (
 	"fmt"
 	"strconv"
+	"time"
 
 	"github.com/ryanhamamura/via/h"
 )
@@ -23,6 +24,8 @@ type triggerOpts struct {
 	value          string
 	window         bool
 	preventDefault bool
+	debounce       time.Duration
+	throttle       time.Duration
 }
 
 type withSignalOpt struct {
@@ -58,6 +61,41 @@ func WithPreventDefault() ActionTriggerOption {
 	return withPreventDefaultOpt{}
 }
 
+type withDebounceOpt struct{ d time.Duration }
+
+func (o withDebounceOpt) apply(opts *triggerOpts) { opts.debounce = o.d }
+
+// WithDebounce adds a debounce modifier to the event trigger.
+func WithDebounce(d time.Duration) ActionTriggerOption { return withDebounceOpt{d} }
+
+type withThrottleOpt struct{ d time.Duration }
+
+func (o withThrottleOpt) apply(opts *triggerOpts) { opts.throttle = o.d }
+
+// WithThrottle adds a throttle modifier to the event trigger.
+func WithThrottle(d time.Duration) ActionTriggerOption { return withThrottleOpt{d} }
+
+// formatDuration renders a duration as e.g. "200ms" for Datastar modifiers.
+func formatDuration(d time.Duration) string {
+	return fmt.Sprintf("%dms", d.Milliseconds())
+}
+
+// buildAttrKey constructs a Datastar attribute key with modifiers.
+// Order: event → debounce/throttle → window.
+func buildAttrKey(event string, opts *triggerOpts) string {
+	key := "on:" + event
+	if opts.debounce > 0 {
+		key += "__debounce." + formatDuration(opts.debounce)
+	}
+	if opts.throttle > 0 {
+		key += "__throttle." + formatDuration(opts.throttle)
+	}
+	if opts.window {
+		key += "__window"
+	}
+	return key
+}
+
 // WithSignal sets a signal value before triggering the action.
 func WithSignal(sig *signal, value string) ActionTriggerOption {
 	return withSignalOpt{
@@ -97,62 +135,62 @@ func actionURL(id string) string {
 // to element nodes in a view.
 func (a *actionTrigger) OnClick(options ...ActionTriggerOption) h.H {
 	opts := applyOptions(options...)
-	return h.Data("on:click", buildOnExpr(actionURL(a.id), &opts))
+	return h.Data(buildAttrKey("click", &opts), buildOnExpr(actionURL(a.id), &opts))
 }
 
 // OnChange returns a via.h DOM attribute that triggers on input change. It can be added
 // to element nodes in a view.
 func (a *actionTrigger) OnChange(options ...ActionTriggerOption) h.H {
 	opts := applyOptions(options...)
-	return h.Data("on:change__debounce.200ms", buildOnExpr(actionURL(a.id), &opts))
+	return h.Data(buildAttrKey("change", &opts), buildOnExpr(actionURL(a.id), &opts))
 }
 
 // OnSubmit returns a via.h DOM attribute that triggers on form submit.
 func (a *actionTrigger) OnSubmit(options ...ActionTriggerOption) h.H {
 	opts := applyOptions(options...)
-	return h.Data("on:submit", buildOnExpr(actionURL(a.id), &opts))
+	return h.Data(buildAttrKey("submit", &opts), buildOnExpr(actionURL(a.id), &opts))
 }
 
 // OnInput returns a via.h DOM attribute that triggers on input (without debounce).
 func (a *actionTrigger) OnInput(options ...ActionTriggerOption) h.H {
 	opts := applyOptions(options...)
-	return h.Data("on:input", buildOnExpr(actionURL(a.id), &opts))
+	return h.Data(buildAttrKey("input", &opts), buildOnExpr(actionURL(a.id), &opts))
 }
 
 // OnFocus returns a via.h DOM attribute that triggers when the element gains focus.
 func (a *actionTrigger) OnFocus(options ...ActionTriggerOption) h.H {
 	opts := applyOptions(options...)
-	return h.Data("on:focus", buildOnExpr(actionURL(a.id), &opts))
+	return h.Data(buildAttrKey("focus", &opts), buildOnExpr(actionURL(a.id), &opts))
 }
 
 // OnBlur returns a via.h DOM attribute that triggers when the element loses focus.
 func (a *actionTrigger) OnBlur(options ...ActionTriggerOption) h.H {
 	opts := applyOptions(options...)
-	return h.Data("on:blur", buildOnExpr(actionURL(a.id), &opts))
+	return h.Data(buildAttrKey("blur", &opts), buildOnExpr(actionURL(a.id), &opts))
 }
 
 // OnMouseEnter returns a via.h DOM attribute that triggers when the mouse enters the element.
 func (a *actionTrigger) OnMouseEnter(options ...ActionTriggerOption) h.H {
 	opts := applyOptions(options...)
-	return h.Data("on:mouseenter", buildOnExpr(actionURL(a.id), &opts))
+	return h.Data(buildAttrKey("mouseenter", &opts), buildOnExpr(actionURL(a.id), &opts))
 }
 
 // OnMouseLeave returns a via.h DOM attribute that triggers when the mouse leaves the element.
 func (a *actionTrigger) OnMouseLeave(options ...ActionTriggerOption) h.H {
 	opts := applyOptions(options...)
-	return h.Data("on:mouseleave", buildOnExpr(actionURL(a.id), &opts))
+	return h.Data(buildAttrKey("mouseleave", &opts), buildOnExpr(actionURL(a.id), &opts))
 }
 
 // OnScroll returns a via.h DOM attribute that triggers on scroll.
 func (a *actionTrigger) OnScroll(options ...ActionTriggerOption) h.H {
 	opts := applyOptions(options...)
-	return h.Data("on:scroll", buildOnExpr(actionURL(a.id), &opts))
+	return h.Data(buildAttrKey("scroll", &opts), buildOnExpr(actionURL(a.id), &opts))
 }
 
 // OnDblClick returns a via.h DOM attribute that triggers on double click.
 func (a *actionTrigger) OnDblClick(options ...ActionTriggerOption) h.H {
 	opts := applyOptions(options...)
-	return h.Data("on:dblclick", buildOnExpr(actionURL(a.id), &opts))
+	return h.Data(buildAttrKey("dblclick", &opts), buildOnExpr(actionURL(a.id), &opts))
 }
 
 // OnKeyDown returns a via.h DOM attribute that triggers when a key is pressed.
@@ -164,11 +202,7 @@ func (a *actionTrigger) OnKeyDown(key string, options ...ActionTriggerOption) h.
 	if key != "" {
 		condition = fmt.Sprintf("evt.key==='%s' &&", key)
 	}
-	attrName := "on:keydown"
-	if opts.window {
-		attrName = "on:keydown__window"
-	}
-	return h.Data(attrName, fmt.Sprintf("%s%s", condition, buildOnExpr(actionURL(a.id), &opts)))
+	return h.Data(buildAttrKey("keydown", &opts), fmt.Sprintf("%s%s", condition, buildOnExpr(actionURL(a.id), &opts)))
 }
 
 // KeyBinding pairs a key with an action and per-binding options.

computed.go

@@ -0,0 +1,55 @@
+package via
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/ryanhamamura/via/h"
+)
+
+// computedSignal is a read-only signal whose value is derived from other signals.
+// It recomputes on every read and is included in patches only when the value changes.
+type computedSignal struct {
+	id      string
+	compute func() string
+	lastVal string
+	changed bool
+}
+
+func (s *computedSignal) ID() string {
+	return s.id
+}
+
+func (s *computedSignal) String() string {
+	return s.compute()
+}
+
+func (s *computedSignal) Int() int {
+	if n, err := strconv.Atoi(s.String()); err == nil {
+		return n
+	}
+	return 0
+}
+
+func (s *computedSignal) Bool() bool {
+	val := strings.ToLower(s.String())
+	return val == "true" || val == "1" || val == "yes" || val == "on"
+}
+
+func (s *computedSignal) Text() h.H {
+	return h.Span(h.Data("text", "$"+s.id))
+}
+
+// recompute calls the compute function and sets changed if the value differs from lastVal.
+func (s *computedSignal) recompute() {
+	val := s.compute()
+	if val != s.lastVal {
+		s.lastVal = val
+		s.changed = true
+	}
+}
+
+func (s *computedSignal) patchValue() string {
+	return fmt.Sprintf("%v", s.lastVal)
+}

computed_test.go

@@ -0,0 +1,190 @@
+package via
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+
+	"github.com/ryanhamamura/via/h"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestComputedBasic(t *testing.T) {
+	v := New()
+	var cs *computedSignal
+	v.Page("/", func(c *Context) {
+		sig1 := c.Signal("hello")
+		sig2 := c.Signal("world")
+		cs = c.Computed(func() string {
+			return sig1.String() + " " + sig2.String()
+		})
+		c.View(func() h.H { return h.Div() })
+	})
+	assert.Equal(t, "hello world", cs.String())
+}
+
+func TestComputedReactivity(t *testing.T) {
+	v := New()
+	var cs *computedSignal
+	var sig1 *signal
+	v.Page("/", func(c *Context) {
+		sig1 = c.Signal("a")
+		sig2 := c.Signal("b")
+		cs = c.Computed(func() string {
+			return sig1.String() + sig2.String()
+		})
+		c.View(func() h.H { return h.Div() })
+	})
+	assert.Equal(t, "ab", cs.String())
+	sig1.SetValue("x")
+	assert.Equal(t, "xb", cs.String())
+}
+
+func TestComputedInt(t *testing.T) {
+	v := New()
+	var cs *computedSignal
+	v.Page("/", func(c *Context) {
+		sig := c.Signal(21)
+		cs = c.Computed(func() string {
+			return fmt.Sprintf("%d", sig.Int()*2)
+		})
+		c.View(func() h.H { return h.Div() })
+	})
+	assert.Equal(t, 42, cs.Int())
+}
+
+func TestComputedBool(t *testing.T) {
+	v := New()
+	var cs *computedSignal
+	v.Page("/", func(c *Context) {
+		sig := c.Signal("true")
+		cs = c.Computed(func() string {
+			return sig.String()
+		})
+		c.View(func() h.H { return h.Div() })
+	})
+	assert.True(t, cs.Bool())
+}
+
+func TestComputedText(t *testing.T) {
+	v := New()
+	var cs *computedSignal
+	v.Page("/", func(c *Context) {
+		cs = c.Computed(func() string { return "hi" })
+		c.View(func() h.H { return h.Div() })
+	})
+
+	var buf bytes.Buffer
+	err := cs.Text().Render(&buf)
+	assert.NoError(t, err)
+	assert.Contains(t, buf.String(), `data-text="$`+cs.ID()+`"`)
+}
+
+func TestComputedChangeDetection(t *testing.T) {
+	v := New()
+	var ctx *Context
+	var sig *signal
+	v.Page("/", func(c *Context) {
+		ctx = c
+		sig = c.Signal("a")
+		c.Computed(func() string {
+			return sig.String() + "!"
+		})
+		c.View(func() h.H { return h.Div() })
+	})
+
+	// First patch includes computed (changed=true from init)
+	patch1 := ctx.prepareSignalsForPatch()
+	assert.NotEmpty(t, patch1)
+
+	// Second patch: nothing changed, computed should not be included
+	patch2 := ctx.prepareSignalsForPatch()
+	// Regular signal still has changed=true (not reset in prepareSignalsForPatch),
+	// but computed should not appear since its value didn't change.
+	hasComputed := false
+	ctx.signals.Range(func(_, value any) bool {
+		if cs, ok := value.(*computedSignal); ok {
+			_, inPatch := patch2[cs.ID()]
+			hasComputed = inPatch
+		}
+		return true
+	})
+	assert.False(t, hasComputed)
+
+	// After changing dependency, computed should reappear
+	sig.SetValue("b")
+	patch3 := ctx.prepareSignalsForPatch()
+	found := false
+	ctx.signals.Range(func(_, value any) bool {
+		if cs, ok := value.(*computedSignal); ok {
+			if v, ok := patch3[cs.ID()]; ok {
+				assert.Equal(t, "b!", v)
+				found = true
+			}
+		}
+		return true
+	})
+	assert.True(t, found)
+}
+
+func TestComputedInComponent(t *testing.T) {
+	v := New()
+	var cs *computedSignal
+	var parentCtx *Context
+	v.Page("/", func(c *Context) {
+		parentCtx = c
+		c.Component(func(comp *Context) {
+			sig := comp.Signal("via")
+			cs = comp.Computed(func() string {
+				return "hello " + sig.String()
+			})
+			comp.View(func() h.H { return h.Div() })
+		})
+		c.View(func() h.H { return h.Div() })
+	})
+
+	assert.Equal(t, "hello via", cs.String())
+	// Verify it's stored on the parent page context
+	found := false
+	parentCtx.signals.Range(func(_, value any) bool {
+		if stored, ok := value.(*computedSignal); ok && stored.ID() == cs.ID() {
+			found = true
+		}
+		return true
+	})
+	assert.True(t, found)
+}
+
+func TestComputedIsReadOnly(t *testing.T) {
+	// Compile-time guarantee: *computedSignal has no Bind() or SetValue() methods.
+	// This test exists as documentation — if someone adds those methods, the
+	// interface assertion below will need updating and serve as a reminder.
+	var cs interface{} = &computedSignal{}
+	type writable interface {
+		SetValue(any)
+	}
+	type bindable interface {
+		Bind() h.H
+	}
+	_, isWritable := cs.(writable)
+	_, isBindable := cs.(bindable)
+	assert.False(t, isWritable, "computedSignal must not have SetValue")
+	assert.False(t, isBindable, "computedSignal must not have Bind")
+}
+
+func TestComputedInjectSignalsSkips(t *testing.T) {
+	v := New()
+	var ctx *Context
+	var cs *computedSignal
+	v.Page("/", func(c *Context) {
+		ctx = c
+		cs = c.Computed(func() string { return "fixed" })
+		c.View(func() h.H { return h.Div() })
+	})
+
+	// Simulate browser sending back a value for the computed signal — should be ignored
+	ctx.injectSignals(map[string]any{
+		cs.ID(): "injected",
+	})
+	assert.Equal(t, "fixed", cs.String())
+}

configuration.go

@@ -53,10 +53,15 @@ type Options struct {
 	// Defaults to "/_datastar.js" if empty.
 	DatastarPath string
 
-	// PubSub enables publish/subscribe messaging. Use vianats.New() for an
-	// embedded NATS backend, or supply any PubSub implementation.
+	// PubSub enables publish/subscribe messaging. When nil, an embedded NATS
+	// server starts automatically in Start(). Supply any PubSub implementation
+	// to replace it.
 	PubSub PubSub
 
+	// Streams declares JetStream streams to create when Start() initializes
+	// the embedded NATS server. Ignored when a custom PubSub is configured.
+	Streams []StreamConfig
+
 	// ContextSuspendAfter is the time a context may be disconnected before
 	// the reaper suspends it (frees page resources but keeps the context
 	// shell for seamless re-init on reconnect). Default: 15m.

context.go

@@ -42,6 +42,7 @@ type Context struct {
 	createdAt           time.Time
 	sseConnected        atomic.Bool
 	sseDisconnectedAt   atomic.Pointer[time.Time]
+	lastSeenAt          atomic.Pointer[time.Time]
 	suspended           atomic.Bool
 }
 
@@ -207,6 +208,40 @@ func (c *Context) Signal(v any) *signal {
 
 }
 
+// Computed creates a read-only signal whose value is derived from the given function.
+// The function is called on every read (String/Int/Bool) for fresh values,
+// and during sync to detect changes for browser patches.
+//
+// Computed signals cannot be bound to inputs or set manually.
+//
+// Example:
+//
+//	full := c.Computed(func() string {
+//	    return first.String() + " " + last.String()
+//	})
+//	c.View(func() h.H {
+//	    return h.Span(full.Text())
+//	})
+func (c *Context) Computed(fn func() string) *computedSignal {
+	sigID := genRandID()
+	initial := fn()
+	cs := &computedSignal{
+		id:      sigID,
+		compute: fn,
+		lastVal: initial,
+		changed: true,
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.isComponent() {
+		c.parentPageCtx.signals.Store(sigID, cs)
+	} else {
+		c.signals.Store(sigID, cs)
+	}
+	return cs
+}
+
 func (c *Context) injectSignals(sigs map[string]any) {
 	if sigs == nil {
 		c.app.logErr(c, "signal injection failed: nil signals")
@@ -248,7 +283,8 @@ func (c *Context) prepareSignalsForPatch() map[string]any {
 	defer c.mu.RUnlock()
 	updatedSigs := make(map[string]any)
 	c.signals.Range(func(sigID, value any) bool {
-		if sig, ok := value.(*signal); ok {
+		switch sig := value.(type) {
+		case *signal:
 			if sig.err != nil {
 				c.app.logWarn(c, "signal '%s' is out of sync: %v", sig.id, sig.err)
 				return true
@@ -256,6 +292,12 @@ func (c *Context) prepareSignalsForPatch() map[string]any {
 			if sig.changed {
 				updatedSigs[sigID.(string)] = fmt.Sprintf("%v", sig.val)
 			}
+		case *computedSignal:
+			sig.recompute()
+			if sig.changed {
+				updatedSigs[sigID.(string)] = sig.patchValue()
+				sig.changed = false
+			}
 		}
 		return true
 	})

docs/pubsub-and-sessions.md

@@ -4,7 +4,7 @@ Infrastructure for multi-user real-time communication and persistent state.
 
 ## PubSub
 
-Via includes an embedded NATS server that starts automatically with `via.New()`. No external services required — pub/sub works out of the box.
+Via includes an embedded NATS server that starts automatically with `v.Start()`. No external services required — pub/sub works out of the box.
 
 ### Interface
 
@@ -73,14 +73,18 @@ This disables the embedded NATS server. The `NATSConn()` and `JetStream()` acces
 
 NATS JetStream provides persistent, replayable message streams. Useful for chat history, event logs, or any scenario where new subscribers need to catch up on past messages.
 
-### Ensure a stream exists
+### Declaring streams
+
+The recommended approach is to declare streams in `Options.Streams`. They are created automatically when `v.Start()` initializes the embedded NATS server:
 
 ```go
-err := via.EnsureStream(v, via.StreamConfig{
+v.Config(via.Options{
+    Streams: []via.StreamConfig{{
         Name:     "CHAT",
         Subjects: []string{"chat.>"},
         MaxMsgs:  1000,
         MaxAge:   24 * time.Hour,
+    }},
 })
 ```
 
@@ -91,7 +95,16 @@ err := via.EnsureStream(v, via.StreamConfig{
 | `MaxMsgs` | Maximum number of messages to retain |
 | `MaxAge` | Maximum age before messages are discarded |
 
-Call `EnsureStream` during app initialization, before `v.Start()`.
+For dynamic stream creation after startup, `EnsureStream` is also available:
+
+```go
+err := via.EnsureStream(v, via.StreamConfig{
+    Name:     "EVENTS",
+    Subjects: []string{"events.>"},
+    MaxMsgs:  500,
+    MaxAge:   12 * time.Hour,
+})
+```
 
 ### Replay history
 

internal/examples/nats-chatroom/README.md

@@ -2,7 +2,7 @@
 
 A chatroom built with Via and an **embedded NATS server**, demonstrating pub/sub messaging as an alternative to the custom `Rooms` implementation in `../chatroom`.
 
-Uses `delaneyj/toolbelt/embeddednats` to run NATS inside the same binary - no external server required.
+Via includes an embedded NATS server that starts automatically — no external server required.
 
 ## Key Differences from Original Chatroom
 
@@ -25,21 +25,6 @@ That's it. No separate NATS server needed.
 
 Open multiple browser tabs at http://localhost:7331 to see messages broadcast across all clients.
 
-## How Embedded NATS Works
-
-```go
-// Start embedded NATS server (JetStream enabled by default)
-ns, err := embeddednats.New(ctx,
-    embeddednats.WithDirectory("./data/nats"),
-)
-ns.WaitForServer()
-
-// Get client connection to embedded server
-nc, err := ns.Client()
-```
-
-Data is persisted to `./data/nats/` for JetStream durability.
-
 ## Architecture
 
 ```
@@ -65,14 +50,16 @@ Data is persisted to `./data/nats/` for JetStream durability.
 
 ## JetStream Durability
 
-Messages persist to disk via JetStream:
+Messages persist to disk via JetStream. Streams are declared in `Options.Streams` and created automatically when `v.Start()` initializes the embedded NATS server:
 
 ```go
-js.AddStream(&nats.StreamConfig{
+v.Config(via.Options{
+    Streams: []via.StreamConfig{{
         Name:     "CHAT",
         Subjects: []string{"chat.>"},
-    MaxMsgs:   1000,  // Keep last 1000 messages
+        MaxMsgs:  1000,
         MaxAge:   24 * time.Hour,
+    }},
 })
 ```
 
@@ -87,23 +74,6 @@ Stop and restart the app - chat history survives.
 - Manual join/leave channels
 
 **This example - ~60 lines of NATS integration:**
-- `embeddednats.New()` starts the server
-- `nc.Subscribe(subject, handler)` for receiving
-- `nc.Publish(subject, data)` for sending
-- NATS handles delivery, no polling
-
-## Next Steps
-
-If this pattern proves useful, it could be promoted to a Via plugin:
-
-```go
-// Hypothetical future API
-v.Config(via.WithEmbeddedNATS("./data/nats"))
-
-// In page init
-c.Subscribe("events.user.*", func(data []byte) {
-    c.Sync()
-})
-
-c.Publish("events.user.login", userData)
-```
+- `via.Subscribe(c, subject, handler)` for receiving
+- `via.Publish(c, subject, data)` for sending
+- Streams declared in `Options` — NATS handles delivery, no polling

internal/examples/nats-chatroom/main.go

@@ -21,17 +21,13 @@ func main() {
 		DocumentTitle: "NATS Chat",
 		LogLevel:      via.LogLevelInfo,
 		ServerAddress: ":7331",
-	})
-
-	err := via.EnsureStream(v, via.StreamConfig{
+		Streams: []via.StreamConfig{{
 			Name:     "CHAT",
 			Subjects: []string{"chat.>"},
 			MaxMsgs:  1000,
 			MaxAge:   24 * time.Hour,
+		}},
 	})
-	if err != nil {
-		log.Fatalf("Failed to ensure stream: %v", err)
-	}
 
 	v.AppendToHead(
 		h.Link(h.Rel("stylesheet"), h.Href("https://cdn.jsdelivr.net/npm/@picocss/pico@2/css/pico.min.css")),
@@ -76,6 +72,6 @@ func main() {
 	protected := v.Group("", requireProfile)
 	protected.Page("/", ChatPage)
 
-	log.Println("Starting NATS chatroom on :7331 (embedded NATS server)")
+	log.Println("Starting NATS chatroom on :7331")
 	v.Start()
 }

internal/examples/nats-chatroom/profile.go

@@ -18,6 +18,12 @@ func ProfilePage(c *via.Context) {
 		via.MaxLen(20, "Must be at most 20 characters"),
 	)
 	selectedEmoji := c.Signal(existingEmoji)
+	previewName := c.Computed(func() string {
+		if name := nameField.String(); name != "" {
+			return name
+		}
+		return "Your Name"
+	})
 
 	saveToSession := func() bool {
 		if !c.ValidateAll() {
@@ -68,18 +74,13 @@ func ProfilePage(c *via.Context) {
 			h.Button(h.Text("Start Chatting"), saveAndChat.OnClick()),
 		)
 
-		previewName := nameField.String()
-		if previewName == "" {
-			previewName = "Your Name"
-		}
-
 		return h.Div(h.Class("profile-page"),
 			h.H2(h.Text("Your Profile"), h.DataViewTransition("page-title")),
 
 			// Live preview
 			h.Div(h.Class("profile-preview"),
 				h.Div(h.Class("avatar avatar-lg"), h.Text(selectedEmoji.String())),
-				h.Span(h.Class("preview-name"), h.Text(previewName)),
+				h.Span(h.Class("preview-name"), previewName.Text()),
 			),
 
 			h.Div(h.Class("profile-form"),

internal/examples/pubsub-crud/main.go

@@ -53,17 +53,13 @@ func main() {
 		DocumentTitle: "Bookmarks",
 		LogLevel:      via.LogLevelInfo,
 		ServerAddress: ":7331",
-	})
-
-	err := via.EnsureStream(v, via.StreamConfig{
+		Streams: []via.StreamConfig{{
 			Name:     "BOOKMARKS",
 			Subjects: []string{"bookmarks.>"},
 			MaxMsgs:  1000,
 			MaxAge:   24 * time.Hour,
+		}},
 	})
-	if err != nil {
-		log.Fatalf("Failed to ensure stream: %v", err)
-	}
 
 	v.AppendToHead(
 		h.Link(h.Rel("stylesheet"), h.Href("https://cdn.jsdelivr.net/npm/daisyui@4/dist/full.min.css")),
@@ -76,6 +72,12 @@ func main() {
 		titleSignal := c.Signal("")
 		urlSignal := c.Signal("")
 		targetIDSignal := c.Signal("")
+		saveLabel := c.Computed(func() string {
+			if targetIDSignal.String() != "" {
+				return "Update Bookmark"
+			}
+			return "Add Bookmark"
+		})
 
 		via.Subscribe(c, "bookmarks.events", func(evt CRUDEvent) {
 			if evt.UserID == userID {
@@ -205,11 +207,6 @@ func main() {
 			}
 			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"),
@@ -225,7 +222,7 @@ func main() {
 					// 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.H2(h.Class("card-title"), saveLabel.Text()),
 							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()),
@@ -233,7 +230,7 @@ func main() {
 									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()),
+									h.Button(h.Class("btn btn-primary"), saveLabel.Text(), save.OnClick()),
 								),
 							),
 						),

nats.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/delaneyj/toolbelt/embeddednats"
+	natsserver "github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 )
 
@@ -50,13 +51,32 @@ func startDefaultNATS() (dn *defaultNATS, err error) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 
-	ns, err := embeddednats.New(ctx, embeddednats.WithDirectory(dataDir))
+	ns, err := embeddednats.New(ctx,
+		embeddednats.WithDirectory(dataDir),
+		embeddednats.WithNATSServerOptions(&natsserver.Options{
+			JetStream: true,
+			StoreDir:  dataDir,
+			Port:      -1,
+		}),
+	)
 	if err != nil {
 		cancel()
 		os.RemoveAll(dataDir)
 		return nil, fmt.Errorf("start embedded nats: %w", err)
 	}
+	ready := make(chan struct{})
+	go func() {
 		ns.WaitForServer()
+		close(ready)
+	}()
+	select {
+	case <-ready:
+	case <-time.After(10 * time.Second):
+		ns.Close()
+		cancel()
+		os.RemoveAll(dataDir)
+		return nil, fmt.Errorf("embedded nats server did not start within 10s")
+	}
 
 	nc, err := ns.Client()
 	if err != nil {

nats_test.go

@@ -10,9 +10,24 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestPubSub_RoundTrip(t *testing.T) {
+// setupNATSTest creates a *V with an embedded NATS server.
+// Skips the test if NATS fails to start (e.g. port conflict in CI).
+func setupNATSTest(t *testing.T) *V {
+	t.Helper()
 	v := New()
-	defer v.Shutdown()
+	dn, err := getSharedNATS()
+	if err != nil {
+		v.Shutdown()
+		t.Skipf("embedded NATS unavailable: %v", err)
+	}
+	v.defaultNATS = dn
+	v.pubsub = &natsRef{dn: dn}
+	t.Cleanup(v.Shutdown)
+	return v
+}
+
+func TestPubSub_RoundTrip(t *testing.T) {
+	v := setupNATSTest(t)
 
 	var received []byte
 	done := make(chan struct{})
@@ -38,8 +53,7 @@ func TestPubSub_RoundTrip(t *testing.T) {
 }
 
 func TestPubSub_MultipleSubscribers(t *testing.T) {
-	v := New()
-	defer v.Shutdown()
+	v := setupNATSTest(t)
 
 	var mu sync.Mutex
 	var results []string
@@ -84,8 +98,7 @@ func TestPubSub_MultipleSubscribers(t *testing.T) {
 }
 
 func TestPubSub_SubscriptionCleanupOnDispose(t *testing.T) {
-	v := New()
-	defer v.Shutdown()
+	v := setupNATSTest(t)
 
 	c := newContext("cleanup-ctx", "/", v)
 	c.View(func() h.H { return h.Div() })
@@ -100,8 +113,7 @@ func TestPubSub_SubscriptionCleanupOnDispose(t *testing.T) {
 }
 
 func TestPubSub_ManualUnsubscribe(t *testing.T) {
-	v := New()
-	defer v.Shutdown()
+	v := setupNATSTest(t)
 
 	c := newContext("unsub-ctx", "/", v)
 	c.View(func() h.H { return h.Div() })
@@ -120,8 +132,7 @@ func TestPubSub_ManualUnsubscribe(t *testing.T) {
 }
 
 func TestPubSub_NoOpDuringPanicCheck(t *testing.T) {
-	v := New()
-	defer v.Shutdown()
+	v := setupNATSTest(t)
 
 	// Panic-check context has id=""
 	c := newContext("", "/", v)

pubsub.go

@@ -1,8 +1,9 @@
 package via
 
 // PubSub is an interface for publish/subscribe messaging backends.
-// By default, New() starts an embedded NATS server. Supply a custom
-// implementation via Config(Options{PubSub: yourBackend}) to override.
+// By default, Start() launches an embedded NATS server if no backend
+// has been configured. Supply a custom implementation via
+// Config(Options{PubSub: yourBackend}) to override.
 type PubSub interface {
 	Publish(subject string, data []byte) error
 	Subscribe(subject string, handler func(data []byte)) (Subscription, error)

pubsub_helpers_test.go

@@ -10,8 +10,7 @@ import (
 )
 
 func TestPublishSubscribe_RoundTrip(t *testing.T) {
-	v := New()
-	defer v.Shutdown()
+	v := setupNATSTest(t)
 
 	type event struct {
 		Name  string `json:"name"`
@@ -43,8 +42,7 @@ func TestPublishSubscribe_RoundTrip(t *testing.T) {
 }
 
 func TestSubscribe_SkipsBadJSON(t *testing.T) {
-	v := New()
-	defer v.Shutdown()
+	v := setupNATSTest(t)
 
 	type msg struct {
 		Text string `json:"text"`

via.go

@@ -145,6 +145,9 @@ func (v *V) Config(cfg Options) {
 	if cfg.ContextTTL != 0 {
 		v.cfg.ContextTTL = cfg.ContextTTL
 	}
+	if cfg.Streams != nil {
+		v.cfg.Streams = cfg.Streams
+	}
 	if cfg.ActionRateLimit.Rate != 0 || cfg.ActionRateLimit.Burst != 0 {
 		v.actionRateLimit = cfg.ActionRateLimit
 	}
@@ -331,15 +334,18 @@ func (v *V) reapOrphanedContexts(suspendAfter, ttl time.Duration) {
 		if c.sseConnected.Load() {
 			continue
 		}
-		var disconnectedFor time.Duration
-		if dc := c.sseDisconnectedAt.Load(); dc != nil {
-			disconnectedFor = now.Sub(*dc)
-		} else {
-			disconnectedFor = now.Sub(c.createdAt)
+		// Use the most recent liveness signal
+		lastAlive := c.createdAt
+		if dc := c.sseDisconnectedAt.Load(); dc != nil && dc.After(lastAlive) {
+			lastAlive = *dc
 		}
-		if disconnectedFor > ttl {
+		if seen := c.lastSeenAt.Load(); seen != nil && seen.After(lastAlive) {
+			lastAlive = *seen
+		}
+		silentFor := now.Sub(lastAlive)
+		if silentFor > ttl {
 			toReap = append(toReap, c)
-		} else if disconnectedFor > suspendAfter && !c.suspended.Load() {
+		} else if silentFor > suspendAfter && !c.suspended.Load() {
 			toSuspend = append(toSuspend, c)
 		}
 	}
@@ -358,6 +364,22 @@ func (v *V) reapOrphanedContexts(suspendAfter, ttl time.Duration) {
 // Start starts the Via HTTP server and blocks until a SIGINT or SIGTERM
 // signal is received, then performs a graceful shutdown.
 func (v *V) Start() {
+	if v.pubsub == nil {
+		dn, err := getSharedNATS()
+		if err != nil {
+			v.logWarn(nil, "embedded NATS unavailable: %v", err)
+		} else {
+			v.defaultNATS = dn
+			v.pubsub = &natsRef{dn: dn}
+		}
+	}
+
+	for _, sc := range v.cfg.Streams {
+		if err := EnsureStream(v, sc); err != nil {
+			v.logger.Fatal().Err(err).Msgf("failed to create stream %q", sc.Name)
+		}
+	}
+
 	handler := http.Handler(v.mux)
 	if v.sessionManager != nil {
 		handler = v.sessionManager.LoadAndSave(v.mux)
@@ -645,6 +667,8 @@ func New() *V {
 			return
 		}
 		c.reqCtx = r.Context()
+		now := time.Now()
+		c.lastSeenAt.Store(&now)
 
 		sse := datastar.NewSSE(w, r, datastar.WithCompression(datastar.WithBrotli(datastar.WithBrotliLevel(5))))
 
@@ -678,17 +702,22 @@ func New() *V {
 
 		go c.Sync()
 
+		keepalive := time.NewTicker(30 * time.Second)
+		defer keepalive.Stop()
+
 		for {
 			select {
 			case <-sse.Context().Done():
 				v.logDebug(c, "SSE connection ended")
 				c.sseConnected.Store(false)
-				now := time.Now()
-				c.sseDisconnectedAt.Store(&now)
+				dcNow := time.Now()
+				c.sseDisconnectedAt.Store(&dcNow)
 				return
 			case <-c.ctxDisposedChan:
 				v.logDebug(c, "context disposed, closing SSE")
 				return
+			case <-keepalive.C:
+				sse.PatchSignals([]byte("{}"))
 			case patch := <-c.patchChan:
 				switch patch.typ {
 				case patchTypeElements:
@@ -833,14 +862,6 @@ func New() *V {
 		v.cleanupCtx(c)
 	})
 
-	dn, err := getSharedNATS()
-	if err != nil {
-		v.logWarn(nil, "embedded NATS unavailable: %v", err)
-	} else {
-		v.defaultNATS = dn
-		v.pubsub = &natsRef{dn: dn}
-	}
-
 	return v
 }
 

via_test.go

@@ -127,7 +127,7 @@ func TestAction(t *testing.T) {
 	v.mux.ServeHTTP(w, req)
 	body := w.Body.String()
 	assert.Contains(t, body, "data-on:click")
-	assert.Contains(t, body, "data-on:change__debounce.200ms")
+	assert.Contains(t, body, "data-on:change")
 	assert.Contains(t, body, "data-on:keydown")
 	assert.Contains(t, body, "/_action/")
 }
@@ -281,6 +281,112 @@ func TestOnKeyDownMap(t *testing.T) {
 	})
 }
 
+func TestFormatDuration(t *testing.T) {
+	tests := []struct {
+		d    time.Duration
+		want string
+	}{
+		{200 * time.Millisecond, "200ms"},
+		{1 * time.Second, "1000ms"},
+		{50 * time.Millisecond, "50ms"},
+	}
+	for _, tt := range tests {
+		assert.Equal(t, tt.want, formatDuration(tt.d))
+	}
+}
+
+func TestBuildAttrKey(t *testing.T) {
+	tests := []struct {
+		name  string
+		event string
+		opts  triggerOpts
+		want  string
+	}{
+		{"bare event", "click", triggerOpts{}, "on:click"},
+		{"debounce only", "change", triggerOpts{debounce: 200 * time.Millisecond}, "on:change__debounce.200ms"},
+		{"throttle only", "scroll", triggerOpts{throttle: 100 * time.Millisecond}, "on:scroll__throttle.100ms"},
+		{"window only", "keydown", triggerOpts{window: true}, "on:keydown__window"},
+		{"debounce + window", "input", triggerOpts{debounce: 300 * time.Millisecond, window: true}, "on:input__debounce.300ms__window"},
+		{"throttle + window", "scroll", triggerOpts{throttle: 500 * time.Millisecond, window: true}, "on:scroll__throttle.500ms__window"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, buildAttrKey(tt.event, &tt.opts))
+		})
+	}
+}
+
+func TestWithDebounce(t *testing.T) {
+	var trigger *actionTrigger
+	v := New()
+	v.Page("/", func(c *Context) {
+		trigger = c.Action(func() {})
+		c.View(func() h.H {
+			return h.Button(trigger.OnClick(WithDebounce(300 * time.Millisecond)))
+		})
+	})
+
+	req := httptest.NewRequest("GET", "/", nil)
+	w := httptest.NewRecorder()
+	v.mux.ServeHTTP(w, req)
+	body := w.Body.String()
+	assert.Contains(t, body, "data-on:click__debounce.300ms")
+	assert.Contains(t, body, "/_action/"+trigger.id)
+}
+
+func TestWithThrottle(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.OnScroll(WithThrottle(100 * time.Millisecond)))
+		})
+	})
+
+	req := httptest.NewRequest("GET", "/", nil)
+	w := httptest.NewRecorder()
+	v.mux.ServeHTTP(w, req)
+	body := w.Body.String()
+	assert.Contains(t, body, "data-on:scroll__throttle.100ms")
+	assert.Contains(t, body, "/_action/"+trigger.id)
+}
+
+func TestWithDebounceOnChange(t *testing.T) {
+	var trigger *actionTrigger
+	v := New()
+	v.Page("/", func(c *Context) {
+		trigger = c.Action(func() {})
+		c.View(func() h.H {
+			return h.Input(trigger.OnChange(WithDebounce(200 * time.Millisecond)))
+		})
+	})
+
+	req := httptest.NewRequest("GET", "/", nil)
+	w := httptest.NewRecorder()
+	v.mux.ServeHTTP(w, req)
+	body := w.Body.String()
+	assert.Contains(t, body, "data-on:change__debounce.200ms")
+	assert.Contains(t, body, "/_action/"+trigger.id)
+}
+
+func TestDebounceWithWindow(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", WithDebounce(150*time.Millisecond), 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__debounce.150ms__window")
+}
+
 func TestConfig(t *testing.T) {
 	v := New()
 	v.Config(Options{DocumentTitle: "Test"})
@@ -312,6 +418,7 @@ func TestReaperCleansOrphanedContexts(t *testing.T) {
 func TestReaperSuspendsContext(t *testing.T) {
 	v := New()
 	c := newContext("suspend-1", "/", v)
+	c.createdAt = time.Now().Add(-30 * time.Minute)
 	dc := time.Now().Add(-20 * time.Minute)
 	c.sseDisconnectedAt.Store(&dc)
 	v.registerCtx(c)
@@ -326,6 +433,7 @@ func TestReaperSuspendsContext(t *testing.T) {
 func TestReaperReapsAfterTTL(t *testing.T) {
 	v := New()
 	c := newContext("reap-1", "/", v)
+	c.createdAt = time.Now().Add(-3 * time.Hour)
 	dc := time.Now().Add(-2 * time.Hour)
 	c.sseDisconnectedAt.Store(&dc)
 	c.suspended.Store(true)
@@ -340,6 +448,7 @@ func TestReaperReapsAfterTTL(t *testing.T) {
 func TestReaperIgnoresAlreadySuspended(t *testing.T) {
 	v := New()
 	c := newContext("already-sus-1", "/", v)
+	c.createdAt = time.Now().Add(-30 * time.Minute)
 	dc := time.Now().Add(-20 * time.Minute)
 	c.sseDisconnectedAt.Store(&dc)
 	c.suspended.Store(true)
@@ -394,6 +503,74 @@ func TestCleanupCtxIdempotent(t *testing.T) {
 	assert.Error(t, err, "context should be removed after cleanup")
 }
 
+func TestReaperRespectsLastSeenAt(t *testing.T) {
+	v := New()
+	c := newContext("seen-1", "/", v)
+	c.createdAt = time.Now().Add(-30 * time.Minute)
+	// Disconnected 20 min ago, but client retried (lastSeenAt) 2 min ago
+	dc := time.Now().Add(-20 * time.Minute)
+	c.sseDisconnectedAt.Store(&dc)
+	seen := time.Now().Add(-2 * time.Minute)
+	c.lastSeenAt.Store(&seen)
+	v.registerCtx(c)
+
+	v.reapOrphanedContexts(15*time.Minute, time.Hour)
+
+	_, err := v.getCtx("seen-1")
+	assert.NoError(t, err, "context with recent lastSeenAt should survive suspend threshold")
+	assert.False(t, c.suspended.Load(), "context should not be suspended")
+}
+
+func TestReaperFallsBackWithoutLastSeenAt(t *testing.T) {
+	v := New()
+	c := newContext("noseen-1", "/", v)
+	c.createdAt = time.Now().Add(-30 * time.Minute)
+	dc := time.Now().Add(-20 * time.Minute)
+	c.sseDisconnectedAt.Store(&dc)
+	// no lastSeenAt set — should fall back to sseDisconnectedAt
+	v.registerCtx(c)
+
+	v.reapOrphanedContexts(15*time.Minute, time.Hour)
+
+	got, err := v.getCtx("noseen-1")
+	assert.NoError(t, err, "context should still be in registry (suspended, not reaped)")
+	assert.True(t, got.suspended.Load(), "context should be suspended using sseDisconnectedAt fallback")
+}
+
+func TestReaperReapsWithStaleLastSeenAt(t *testing.T) {
+	v := New()
+	c := newContext("stale-seen-1", "/", v)
+	c.createdAt = time.Now().Add(-3 * time.Hour)
+	dc := time.Now().Add(-2 * time.Hour)
+	c.sseDisconnectedAt.Store(&dc)
+	// lastSeenAt is also old — beyond TTL
+	seen := time.Now().Add(-90 * time.Minute)
+	c.lastSeenAt.Store(&seen)
+	c.suspended.Store(true)
+	v.registerCtx(c)
+
+	v.reapOrphanedContexts(15*time.Minute, time.Hour)
+
+	_, err := v.getCtx("stale-seen-1")
+	assert.Error(t, err, "context with stale lastSeenAt beyond TTL should be reaped")
+}
+
+func TestLastSeenAtUpdatedOnSSEConnect(t *testing.T) {
+	v := New()
+	c := newContext("seen-sse-1", "/", v)
+	v.registerCtx(c)
+
+	assert.Nil(t, c.lastSeenAt.Load(), "lastSeenAt should be nil before SSE connect")
+
+	// Simulate what the SSE handler does after getCtx
+	now := time.Now()
+	c.lastSeenAt.Store(&now)
+
+	got := c.lastSeenAt.Load()
+	assert.NotNil(t, got, "lastSeenAt should be set after SSE connect")
+	assert.WithinDuration(t, now, *got, time.Second)
+}
+
 func TestDevModeRemovePersistedFix(t *testing.T) {
 	v := New()
 	v.cfg.DevMode = true