host

func TestTypedActionRejectsUnknownFields(t *testing.T)

{
	type request struct {
		Name string `json:"name"`
	}
	type response struct {
		Greeting string `json:"greeting"`
	}
	const name = "test.typed.strict"
	if err := RegisterAction(name, func(_ context.Context, _ *Session, request request) (response, error) {
		return response{Greeting: "hello " + request.Name}, nil
	}); err != nil {
		t.Fatalf("register action: %v", err)
	}

	result, actionErr := DispatchAction(context.Background(), newSession("typed"), name, map[string]any{"name": "Ada"})
	if actionErr != nil {
		t.Fatalf("dispatch action: %v", actionErr)
	}
	if result.(response).Greeting != "hello Ada" {
		t.Fatalf("unexpected response: %#v", result)
	}

	_, actionErr = DispatchAction(context.Background(), newSession("strict"), name, map[string]any{
		"name":  "Ada",
		"admin": true,
	})
	if actionErr == nil || actionErr.Code != "invalid_request" {
		t.Fatalf("unknown field was accepted: %#v", actionErr)
	}
}

func TestTypedActionAuthorizationHidesInternalError(t *testing.T)

{
	type request struct {
		Owner string `json:"owner"`
	}
	const name = "test.typed.authorized"
	if err := RegisterAction(name,
		func(_ context.Context, _ *Session, request request) (request, error) {
			return request, nil
		},
		WithActionAuthorizer(func(_ context.Context, _ *Session, request request) error {
			if request.Owner != "allowed" {
				return errors.New("database policy detail")
			}
			return nil
		}),
	); err != nil {
		t.Fatalf("register action: %v", err)
	}

	_, actionErr := DispatchAction(context.Background(), newSession("denied"), name, map[string]any{"owner": "denied"})
	if actionErr == nil || actionErr.Code != "forbidden" || actionErr.Message != "action forbidden" {
		t.Fatalf("unexpected authorization response: %#v", actionErr)
	}
}

func TestTypedFormReturnsFieldErrors(t *testing.T)

{
	type values struct {
		Email string `json:"email"`
	}
	type result struct {
		ID int `json:"id"`
	}
	const name = "test.form.validation"
	if err := RegisterForm(name,
		func(values values) FieldErrors {
			if values.Email == "" {
				return FieldErrors{"email": "required"}
			}
			return nil
		},
		func(_ context.Context, _ *Session, _ values) (result, error) {
			return result{ID: 7}, nil
		},
	); err != nil {
		t.Fatalf("register form: %v", err)
	}

	raw, actionErr := DispatchAction(context.Background(), newSession("form"), name, map[string]any{})
	if actionErr != nil {
		t.Fatalf("dispatch form: %v", actionErr)
	}
	response := raw.(FormResponse[result])
	if response.Valid || response.Fields["email"] != "required" {
		t.Fatalf("unexpected form response: %#v", response)
	}
}

func TestClientCanUnsubscribeFromBroadcasts(t *testing.T)

{
	const componentName = "broadcast-unsubscribe"
	Register(NewHostComponent(componentName, func(map[string]any) any {
		return map[string]any{"ready": true}
	}))

	root := t.TempDir()
	if err := os.WriteFile(filepath.Join(root, "index.html"), []byte("ok"), 0o600); err != nil {
		t.Fatalf("write index: %v", err)
	}
	srv := httptest.NewServer(NewMux(root))
	defer srv.Close()
	ws, err := websocket.Dial("ws"+strings.TrimPrefix(srv.URL, "http")+"/ws", "", srv.URL)
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	defer closeTestResource(t, ws)

	init, _ := json.Marshal(map[string]any{"component": componentName, "payload": map[string]any{"init": true}})
	if err := websocket.Message.Send(ws, init); err != nil {
		t.Fatalf("subscribe: %v", err)
	}
	var raw []byte
	if err := websocket.Message.Receive(ws, &raw); err != nil {
		t.Fatalf("receive init: %v", err)
	}

	unsubscribe, _ := json.Marshal(map[string]any{"component": componentName, "payload": map[string]any{"unsubscribe": true}})
	if err := websocket.Message.Send(ws, unsubscribe); err != nil {
		t.Fatalf("unsubscribe: %v", err)
	}
	if err := websocket.Message.Receive(ws, &raw); err != nil {
		t.Fatalf("receive unsubscribe acknowledgement: %v", err)
	}

	Broadcast(componentName, map[string]any{"unexpected": true})
	if err := ws.SetDeadline(time.Now().Add(50 * time.Millisecond)); err != nil {
		t.Fatalf("set deadline: %v", err)
	}
	if err := websocket.Message.Receive(ws, &raw); err == nil {
		t.Fatalf("received broadcast after unsubscribe: %s", raw)
	}
	if err := ws.SetDeadline(time.Time{}); err != nil {
		t.Fatalf("reset deadline: %v", err)
	}
}

func TestBroadcastConcurrentWithConnectionChurn(t *testing.T)

{
	const componentName = "broadcast-churn"
	Register(NewHostComponentWithSession(componentName, func(_ *Session, _ map[string]any) any {
		return map[string]any{"ok": true}
	}))

	root := t.TempDir()
	if err := os.WriteFile(filepath.Join(root, "index.html"), []byte("ok"), 0o600); err != nil {
		t.Fatalf("write index: %v", err)
	}
	srv := httptest.NewServer(NewMux(root))
	defer srv.Close()
	wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "/ws"

	init, err := json.Marshal(map[string]any{
		"component": componentName,
		"payload":   map[string]any{"init": true},
	})
	if err != nil {
		t.Fatalf("marshal init: %v", err)
	}

	var wg sync.WaitGroup
	// Churn: connections subscribe, receive the handler response and close.
	for i := 0; i < 8; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := 0; j < 5; j++ {
				ws, err := websocket.Dial(wsURL, "", srv.URL)
				if err != nil {
					continue
				}
				if err := websocket.Message.Send(ws, init); err == nil {
					var raw []byte
					_ = websocket.Message.Receive(ws, &raw)
				}
				closeTestResource(t, ws)
			}
		}()
	}
	// Broadcasters run against the same component while the map churns.
	for i := 0; i < 4; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := 0; j < 50; j++ {
				Broadcast(componentName, map[string]any{"tick": j})
			}
		}()
	}
	wg.Wait()
}

func TestNewMuxDebugEndpoints(t *testing.T)

{
	t.Setenv("RFW_DEVTOOLS", "1")
	mux := NewMux(t.TempDir())
	ts := httptest.NewServer(mux)
	defer ts.Close()

	resp, err := http.Get(ts.URL + "/debug/vars")
	if err != nil {
		t.Fatalf("vars request failed: %v", err)
	}
	if err := resp.Body.Close(); err != nil {
		t.Fatalf("close debug response: %v", err)
	}
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("expected 200, got %d", resp.StatusCode)
	}

	resp, err = http.Get(ts.URL + "/debug/pprof/")
	if err != nil {
		t.Fatalf("pprof request failed: %v", err)
	}
	if err := resp.Body.Close(); err != nil {
		t.Fatalf("close pprof response: %v", err)
	}
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("expected 200, got %d", resp.StatusCode)
	}
}

func TestNewMuxFSServesEmbeddedBuild(t *testing.T)

{
	srv := httptest.NewServer(NewMuxFS(testClientFS()))
	defer srv.Close()

	type result struct {
		status int
		header http.Header
	}
	// get issues the request and closes the body before returning, so the
	// assertions never hold an open response.
	get := func(path, accept string) result {
		req, err := http.NewRequest(http.MethodGet, srv.URL+path, nil)
		if err != nil {
			t.Fatalf("new request %s: %v", path, err)
		}
		if accept != "" {
			req.Header.Set("Accept", accept)
		}
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			t.Fatalf("get %s: %v", path, err)
		}
		if cerr := resp.Body.Close(); cerr != nil {
			t.Fatalf("close body %s: %v", path, cerr)
		}
		return result{status: resp.StatusCode, header: resp.Header}
	}

	if got := get("/", "text/html").status; got != http.StatusOK {
		t.Fatalf("root status = %d, want 200", got)
	}
	if got := get("/assets/app.css", "").status; got != http.StatusOK {
		t.Fatalf("nested asset status = %d, want 200", got)
	}
	if got := get("/app.wasm?v=abc", "").header.Get("Cache-Control"); got != "public, max-age=31536000, immutable" {
		t.Fatalf("versioned wasm Cache-Control = %q", got)
	}
	if got := get("/app.wasm.br", "").header.Get("Content-Encoding"); got != "br" {
		t.Fatalf("wasm.br Content-Encoding = %q, want br", got)
	}
	if got := get("/rfw_config.js", "").header.Get("Cache-Control"); got != "no-cache" {
		t.Fatalf("rfw_config.js Cache-Control = %q, want no-cache", got)
	}
	// An unknown HTML route falls back to index.html (single-page app).
	if got := get("/dashboard/live", "text/html").status; got != http.StatusOK {
		t.Fatalf("unknown html route status = %d, want 200 (index fallback)", got)
	}
	// A missing non-HTML asset is a 404, not the index.
	if got := get("/missing.css", "").status; got != http.StatusNotFound {
		t.Fatalf("missing asset status = %d, want 404", got)
	}
	// A plain GET is not a WebSocket handshake, so /ws rejects it, but it must be
	// routed rather than falling through to the catch-all 404.
	if got := get("/ws", "").status; got == http.StatusNotFound {
		t.Fatalf("/ws returned 404, endpoint not registered")
	}
}

func wsProbe(t *testing.T, mux *http.ServeMux, origin string) int

{
	t.Helper()
	srv := httptest.NewServer(mux)
	defer srv.Close()
	req, err := http.NewRequest(http.MethodGet, srv.URL+"/ws", nil)
	if err != nil {
		t.Fatalf("request: %v", err)
	}
	if origin != "" {
		req.Header.Set("Origin", origin)
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("do: %v", err)
	}
	if err := resp.Body.Close(); err != nil {
		t.Fatalf("close response: %v", err)
	}
	return resp.StatusCode
}

func TestWSOriginAllowlist(t *testing.T)

{
	mux := NewMux(t.TempDir(), WithOriginAllowlist("https://app.example.com"))
	if code := wsProbe(t, mux, "http://evil.example"); code != http.StatusForbidden {
		t.Fatalf("expected 403 for unlisted origin, got %d", code)
	}
	if code := wsProbe(t, mux, ""); code != http.StatusForbidden {
		t.Fatalf("expected 403 for missing origin, got %d", code)
	}
	if code := wsProbe(t, mux, "https://app.example.com"); code == http.StatusForbidden {
		t.Fatalf("allowed origin rejected: %d", code)
	}
}

func TestWSAuthFunc(t *testing.T)

{
	mux := NewMux(t.TempDir(), WithAuthFunc(func(r *http.Request) bool {
		return r.Header.Get("Origin") == "https://trusted.example.com"
	}))
	if code := wsProbe(t, mux, "http://evil.example"); code != http.StatusUnauthorized {
		t.Fatalf("expected 401 for rejected auth, got %d", code)
	}
	if code := wsProbe(t, mux, "https://trusted.example.com"); code == http.StatusUnauthorized {
		t.Fatalf("accepted auth rejected: %d", code)
	}
}

func TestWSConnectionLimit(t *testing.T)

{
	runtime := NewWSRuntime(WithSSCLimits(SSCLimits{MaxConnections: 1}))
	if !runtime.AcquireConnection() {
		t.Fatal("first connection was rejected")
	}
	if runtime.AcquireConnection() {
		t.Fatal("second connection exceeded the limit")
	}
	runtime.ReleaseConnection()
	if !runtime.AcquireConnection() {
		t.Fatal("released connection slot was not reusable")
	}
	runtime.ReleaseConnection()
}

func TestNewMuxServesBrotliWasmWithEncoding(t *testing.T)

{
	t.Setenv("RFW_DEVTOOLS", "")
	root := t.TempDir()
	clientDir := filepath.Join(root, "client")
	if err := os.MkdirAll(clientDir, 0o755); err != nil {
		t.Fatalf("failed to create client dir: %v", err)
	}
	wasmPath := filepath.Join(clientDir, "app.wasm.br")
	if err := os.WriteFile(wasmPath, []byte("compressed"), 0o644); err != nil {
		t.Fatalf("failed to write wasm: %v", err)
	}
	if err := os.WriteFile(filepath.Join(clientDir, "index.html"), []byte("<html></html>"), 0o644); err != nil {
		t.Fatalf("failed to write index: %v", err)
	}
	if err := os.WriteFile(filepath.Join(clientDir, "rfw_config.js"), []byte("//cfg"), 0o644); err != nil {
		t.Fatalf("failed to write config: %v", err)
	}

	mux := NewMux(clientDir)
	srv := httptest.NewServer(mux)
	defer srv.Close()

	resp, err := http.Get(srv.URL + "/app.wasm.br?v=abc123")
	if err != nil {
		t.Fatalf("failed to get wasm: %v", err)
	}
	defer closeTestResource(t, resp.Body)

	if resp.StatusCode != http.StatusOK {
		t.Fatalf("unexpected status: %d", resp.StatusCode)
	}
	if enc := resp.Header.Get("Content-Encoding"); enc != "br" {
		t.Fatalf("expected Content-Encoding br, got %q", enc)
	}
	if ct := resp.Header.Get("Content-Type"); ct != "application/wasm" {
		t.Fatalf("expected Content-Type application/wasm, got %q", ct)
	}
	if cache := resp.Header.Get("Cache-Control"); cache != "public, max-age=31536000, immutable" {
		t.Fatalf("unexpected Cache-Control header: %q", cache)
	}
	if vary := resp.Header.Get("Vary"); vary != "Accept-Encoding" && vary != "Accept-Encoding, Accept-Encoding" {
		// Allow duplicated value as Go's header may append values depending on environment.
		t.Fatalf("unexpected Vary header: %q", vary)
	}
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		t.Fatalf("failed to read body: %v", err)
	}
	if string(body) != "compressed" {
		t.Fatalf("unexpected body: %q", string(body))
	}

	resp, err = http.Get(srv.URL + "/app.wasm.br?v=")
	if err != nil {
		t.Fatalf("failed to get unversioned wasm: %v", err)
	}
	defer closeTestResource(t, resp.Body)
	if cache := resp.Header.Get("Cache-Control"); cache != "no-cache" {
		t.Fatalf("unexpected unversioned Cache-Control header: %q", cache)
	}

	resp, err = http.Get(srv.URL + "/rfw_config.js")
	if err != nil {
		t.Fatalf("failed to get runtime config: %v", err)
	}
	defer closeTestResource(t, resp.Body)
	if cache := resp.Header.Get("Cache-Control"); cache != "no-cache" {
		t.Fatalf("unexpected runtime config Cache-Control header: %q", cache)
	}
}

func TestNewMuxDevModeNoStore(t *testing.T)

{
	t.Setenv("RFW_DEVTOOLS", "")
	t.Setenv("RFW_DEV_BUILD", "1")
	root := t.TempDir()
	clientDir := filepath.Join(root, "client")
	if err := os.MkdirAll(clientDir, 0o755); err != nil {
		t.Fatalf("failed to create client dir: %v", err)
	}
	if err := os.WriteFile(filepath.Join(clientDir, "app.wasm"), []byte("wasm"), 0o644); err != nil {
		t.Fatalf("failed to write wasm: %v", err)
	}
	if err := os.WriteFile(filepath.Join(clientDir, "rfw_config.js"), []byte("//cfg"), 0o644); err != nil {
		t.Fatalf("failed to write config: %v", err)
	}
	if err := os.WriteFile(filepath.Join(clientDir, "index.html"), []byte("<html></html>"), 0o644); err != nil {
		t.Fatalf("failed to write index: %v", err)
	}

	mux := NewMux(clientDir)
	srv := httptest.NewServer(mux)
	defer srv.Close()

	for _, path := range []string{"/app.wasm?v=abc123", "/rfw_config.js", "/"} {
		resp, err := http.Get(srv.URL + path)
		if err != nil {
			t.Fatalf("get %s: %v", path, err)
		}
		cache := resp.Header.Get("Cache-Control")
		closeTestResource(t, resp.Body)
		if cache != "no-store" {
			t.Fatalf("dev %s: expected no-store, got %q", path, cache)
		}
	}
}

func newSession(id string, options ...sessionOptions) *Session

{
	var config sessionOptions
	if len(options) > 0 {
		config = options[0]
	}
	return &Session{
		id:          id,
		resumeToken: config.resumeToken,
		stores:      state.NewStoreManager(),
		ctx:         make(map[string]any),
		attached:    true,
		replayLimit: config.replayLimit,
	}
}

func allocateSession(replayLimit, maxSessions int) (*Session, error)

{
	id := generateSessionID()
	token := ""
	if replayLimit > 0 {
		token = generateSessionID() + generateSessionID()
	}
	session := newSession(id, sessionOptions{resumeToken: token, replayLimit: replayLimit})
	sessionMu.Lock()
	if maxSessions > 0 && len(sessions) >= maxSessions {
		sessionMu.Unlock()
		return nil, ErrSessionLimit
	}
	sessions[id] = session
	if token != "" {
		sessionByToken[token] = session
	}
	sessionMu.Unlock()
	return session, nil
}

func SuspendSession(session *Session, ttl time.Duration)

{
	if session == nil {
		return
	}
	session.outboundMu.Lock()
	session.deliveryMu.Lock()
	if !session.attached {
		session.deliveryMu.Unlock()
		session.outboundMu.Unlock()
		return
	}
	session.connection = nil
	session.attached = false
	if ttl <= 0 || session.resumeToken == "" {
		session.deliveryMu.Unlock()
		session.outboundMu.Unlock()
		ReleaseSession(session)
		return
	}
	expires := time.Now().Add(ttl)
	session.expires = expires
	session.expiryTimer = time.AfterFunc(ttl, func() {
		releaseSession(session, expires)
	})
	session.deliveryMu.Unlock()
	session.outboundMu.Unlock()
}

func ResumeSession(token string) (*Session, bool)

{
	if token == "" {
		return nil, false
	}
	sessionMu.RLock()
	session := sessionByToken[token]
	sessionMu.RUnlock()
	if session == nil {
		return nil, false
	}
	session.deliveryMu.Lock()
	defer session.deliveryMu.Unlock()
	if session.released || session.attached || (!session.expires.IsZero() && time.Now().After(session.expires)) {
		return nil, false
	}
	session.attached = true
	session.resumePending = true
	session.expires = time.Time{}
	if session.expiryTimer != nil {
		session.expiryTimer.Stop()
		session.expiryTimer = nil
	}
	return session, true
}

func releaseSession(session *Session, expectedExpiry time.Time)

{
	if session == nil {
		return
	}
	session.outboundMu.Lock()
	session.deliveryMu.Lock()
	if session.released || (!expectedExpiry.IsZero() &&
		(session.attached || !session.expires.Equal(expectedExpiry))) {
		session.deliveryMu.Unlock()
		session.outboundMu.Unlock()
		return
	}
	session.released = true
	if session.expiryTimer != nil {
		session.expiryTimer.Stop()
		session.expiryTimer = nil
	}
	session.connection = nil
	session.connectionManaged = false
	session.resumePending = false
	session.attached = false
	session.deliveryMu.Unlock()
	session.outboundMu.Unlock()
	sessionMu.Lock()
	if sessions[session.id] == session {
		delete(sessions, session.id)
	}
	if session.resumeToken != "" && sessionByToken[session.resumeToken] == session {
		delete(sessionByToken, session.resumeToken)
	}
	sessionMu.Unlock()
}

func validTagName(tag string) bool

{
	if tag == "" || !isASCIILetter(rune(tag[0])) {
		return false
	}
	for _, char := range tag {
		if isASCIILetter(char) || char >= '0' && char <= '9' || char == '-' {
			continue
		}
		return false
	}
	return true
}

func hostVarTag(tag, name string, value any, escape bool) string

{
	if !validTagName(tag) {
		return ""
	}
	v := fmt.Sprintf("%v", value)
	body := v
	expected := ""
	if escape {
		body = html.EscapeString(v)
		expected = html.EscapeString(v)
	}
	return fmt.Sprintf(`<%s %s="%s" %s="%s">%s</%s>`,
		tag, hostVarAttr, html.EscapeString(name), hostExpectedAttr, expected, body, tag)
}

func TestSpan(t *testing.T)

{
	got := Span("Visit", 1)
	if !strings.Contains(got, `data-host-var="Visit"`) {
		t.Fatalf("missing data-host-var: %s", got)
	}
	if !strings.Contains(got, `data-host-expected="1"`) {
		t.Fatalf("missing data-host-expected: %s", got)
	}
	if !strings.Contains(got, ">1</span>") {
		t.Fatalf("missing value: %s", got)
	}
	if !strings.HasPrefix(got, "<span ") {
		t.Fatalf("wrong tag: %s", got)
	}
}

func TestHelpersEscapeValues(t *testing.T)

{
	got := Span("msg", `<img src=x onerror=alert(1)>`)
	if !strings.Contains(got, "&lt;img src=x onerror=alert(1)&gt;") {
		t.Fatalf("value not escaped: %s", got)
	}
	if strings.Contains(got, "><img") {
		t.Fatalf("markup injected: %s", got)
	}
	if !strings.Contains(got, `data-host-expected="&lt;img src=x onerror=alert(1)&gt;"`) {
		t.Fatalf("expected escaped migration value: %s", got)
	}
}

func ResolveRoot(root string) string

{
	if _, err := os.Stat(root); err == nil {
		return root
	}
	if exe, err := os.Executable(); err == nil {
		candidate := filepath.Join(filepath.Dir(exe), "..", root)
		if _, err := os.Stat(candidate); err == nil {
			return candidate
		}
	}
	return root
}

func NewMux(root string, opts ...MuxOption) *http.ServeMux

{
	root = ResolveRoot(root)
	runtime := NewWSRuntime(opts...)
	staticRoot := filepath.Join(root, "..", "static")
	mux := http.NewServeMux()
	if os.Getenv("RFW_DEVTOOLS") != "" {
		mux.Handle("/debug/vars", expvar.Handler())
		mux.HandleFunc("/debug/pprof/", pprof.Index)
		mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
		mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
		mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
		mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
	}
	fs := http.FileServer(http.Dir(root))
	rootDir := http.Dir(root)
	var sfs http.Handler
	var staticDir http.Dir
	if _, err := os.Stat(staticRoot); err == nil {
		staticDir = http.Dir(staticRoot)
		sfs = http.FileServer(staticDir)
	}
	if sfs != nil {
		mux.Handle("/static/", http.StripPrefix("/static", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			setWasmEncodingHeaders(w, r.URL.Path, r.URL.Query().Get("v") != "")
			sfs.ServeHTTP(w, r)
		})))
	}
	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		if sfs != nil {
			if regularFile(staticDir, r.URL.Path) {
				setWasmEncodingHeaders(w, r.URL.Path, r.URL.Query().Get("v") != "")
				sfs.ServeHTTP(w, r)
				return
			}
		}
		if regularFile(rootDir, r.URL.Path) {
			setWasmEncodingHeaders(w, r.URL.Path, r.URL.Query().Get("v") != "")
			fs.ServeHTTP(w, r)
			return
		}
		// Serve index.html only for HTML requests or bare paths to avoid
		// returning HTML for CSS, JS, image, etc. requests.
		accept := r.Header.Get("Accept")
		if strings.Contains(accept, "text/html") || r.URL.Path == "/" || r.URL.Path == "" {
			if devMode() {
				w.Header().Set("Cache-Control", "no-store")
			}
			http.ServeFile(w, r, filepath.Join(root, "index.html"))
			return
		}
		http.NotFound(w, r)
	})
	mux.Handle("/ws", runtime.Guard(websocket.Handler(func(ws *websocket.Conn) {
		wsHandler(ws, runtime)
	})))
	return mux
}

func setWasmEncodingHeaders(w http.ResponseWriter, path string, versioned bool)

{
	// In dev, nothing may be cached: the wasm version pointer lives in
	// rfw_config.js and the binary is fetched as app.wasm?v=<hash>. Caching
	// either one leaves the browser re-requesting a stale ?v= against an
	// immutable entry, so rebuilds are never picked up. no-store on every asset
	// forces a fresh fetch each load. Production keeps the immutable policy.
	if devMode() {
		w.Header().Set("Cache-Control", "no-store")
	} else if strings.Trim(path, "/") == "rfw_config.js" {
		// This small file points the loader at the content-versioned WASM URL.
		// It must revalidate across deployments so an old pointer cannot keep a
		// browser on an otherwise correctly immutable old binary.
		w.Header().Set("Cache-Control", "no-cache")
	}
	if !strings.HasSuffix(path, ".wasm") && !strings.HasSuffix(path, ".wasm.br") {
		return
	}
	header := w.Header()
	if !devMode() {
		if versioned {
			header.Set("Cache-Control", "public, max-age=31536000, immutable")
		} else {
			header.Set("Cache-Control", "no-cache")
		}
	}
	if !strings.HasSuffix(path, ".wasm.br") {
		return
	}
	header.Set("Content-Encoding", "br")
	header.Set("Content-Type", "application/wasm")
	if vary := header.Get("Vary"); vary == "" {
		header.Set("Vary", "Accept-Encoding")
	} else if !strings.Contains(vary, "Accept-Encoding") {
		header.Set("Vary", vary+", Accept-Encoding")
	}
}

func generateSelfSignedCert() (tls.Certificate, error)

{
	priv, err := rsa.GenerateKey(rand.Reader, 2048)
	if err != nil {
		return tls.Certificate{}, err
	}
	tmpl := x509.Certificate{
		SerialNumber: big.NewInt(1),
		NotBefore:    time.Now(),
		NotAfter:     time.Now().Add(365 * 24 * time.Hour),
		KeyUsage:     x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
		DNSNames:     []string{"localhost"},
	}
	der, err := x509.CreateCertificate(rand.Reader, &tmpl, &tmpl, &priv.PublicKey, priv)
	if err != nil {
		return tls.Certificate{}, err
	}
	certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
	keyPEM := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(priv)})
	return tls.X509KeyPair(certPEM, keyPEM)
}

func TestSessionResumeAndReplay(t *testing.T)

{
	session := AllocateResumableSession(4)
	token := session.ResumeToken()
	first := session.PrepareOutbound(Outbound{Component: "Counter", Payload: map[string]any{"value": 1}})
	second := session.PrepareOutbound(Outbound{Component: "Counter", Payload: map[string]any{"value": 2}})
	if first.Sequence != 1 || second.Sequence != 2 || token == "" {
		t.Fatalf("missing delivery metadata: first=%#v second=%#v", first, second)
	}

	SuspendSession(session, time.Second)
	resumed, ok := ResumeSession(token)
	if !ok || resumed != session {
		t.Fatal("session did not resume")
	}
	replay, err := resumed.ReplayAfter(first.Sequence)
	if err != nil {
		t.Fatalf("replay: %v", err)
	}
	if len(replay) != 1 || replay[0].Sequence != second.Sequence {
		t.Fatalf("unexpected replay: %#v", replay)
	}
	ReleaseSession(session)
}

func TestSessionRejectsDuplicatesAndGaps(t *testing.T)

{
	session := newSession("ordered")
	if err := session.AcceptInbound(1); err != nil {
		t.Fatalf("first message: %v", err)
	}
	if err := session.AcceptInbound(1); !errors.Is(err, ErrDuplicateMessage) {
		t.Fatalf("duplicate result: %v", err)
	}
	if err := session.AcceptInbound(3); !errors.Is(err, ErrSequenceGap) {
		t.Fatalf("gap result: %v", err)
	}
	if err := session.AcceptInbound(2); err != nil {
		t.Fatalf("next message: %v", err)
	}
}

func TestSessionAllocationLimit(t *testing.T)

{
	sessionMu.RLock()
	current := len(sessions)
	sessionMu.RUnlock()

	session, err := allocateSession(1, current+1)
	if err != nil {
		t.Fatalf("allocate within limit: %v", err)
	}
	defer ReleaseSession(session)
	if _, err := allocateSession(1, current+1); !errors.Is(err, ErrSessionLimit) {
		t.Fatalf("expected session limit, got %v", err)
	}
}

func TestWithoutSSCResumeCreatesEphemeralSession(t *testing.T)

{
	runtime := NewWSRuntime(WithoutSSCResume())
	session, err := runtime.NewSession(nil)
	if err != nil {
		t.Fatalf("new session: %v", err)
	}
	if session.ResumeToken() != "" || runtime.ResumeTTL() != 0 {
		t.Fatalf("resume was not disabled: token=%q ttl=%s", session.ResumeToken(), runtime.ResumeTTL())
	}
	ReleaseSession(session)
}

func TestExpiredTimerDoesNotReleaseResumedSession(t *testing.T)

{
	session := AllocateResumableSession(1)
	defer ReleaseSession(session)
	token := session.ResumeToken()
	SuspendSession(session, time.Second)
	session.deliveryMu.Lock()
	expectedExpiry := session.expires
	session.deliveryMu.Unlock()
	if _, ok := ResumeSession(token); !ok {
		t.Fatal("session did not resume")
	}

	releaseSession(session, expectedExpiry)

	if current, ok := SessionByID(session.ID()); !ok || current != session {
		t.Fatal("stale expiry released the resumed session")
	}
}

func readPort() int

{
	if override := strings.TrimSpace(os.Getenv("RFW_HOST_PORT")); override != "" {
		if p, err := strconv.Atoi(override); err == nil && p > 0 {
			return p
		}
	}
	var manifest struct {
		Port int `json:"port"`
	}
	data, err := os.ReadFile("rfw.json")
	if err != nil {
		return 8080
	}
	if err := json.Unmarshal(data, &manifest); err != nil {
		return 8080
	}
	if manifest.Port == 0 {
		return 8080
	}
	return manifest.Port
}

func resolveRoot() string

{
	// Check rfw.json for build configuration.
	var manifest struct {
		Build struct {
			Dir string `json:"dir"`
		} `json:"build"`
	}
	if data, err := os.ReadFile("rfw.json"); err == nil {
		_ = json.Unmarshal(data, &manifest)
		if manifest.Build.Dir != "" {
			return manifest.Build.Dir
		}
	}
	return "build/client"
}

func Start(root string) error

{
	port := readPort()
	httpsPort := port + 1

	go func() {
		addr := fmt.Sprintf(":%d", port)
		if err := ListenAndServe(addr, root); err != nil {
			logger.Error("HTTP server error", "err", err)
		}
	}()

	httpsAddr := fmt.Sprintf(":%d", httpsPort)
	return ListenAndServeTLS(httpsAddr, root)
}

func RegisterAction[Request, Response any](name string, handler ActionHandler[Request, Response], opts ...ActionOption[Request]) error

{
	if name == "" {
		return errors.New("host: empty action name")
	}
	if handler == nil {
		return errors.New("host: nil action handler")
	}
	var config actionConfig[Request]
	for _, opt := range opts {
		opt(&config)
	}
	actionRegistry.Lock()
	defer actionRegistry.Unlock()
	if _, exists := actionRegistry.actions[name]; exists {
		return fmt.Errorf("host: action %q already registered", name)
	}
	actionRegistry.actions[name] = typedAction[Request, Response]{
		handler:   handler,
		authorize: config.authorize,
	}
	return nil
}

func decodeActionPayload(payload map[string]any, target any) error

{
	if payload == nil {
		payload = map[string]any{}
	}
	data, err := json.Marshal(payload)
	if err != nil {
		return errors.New("request payload is not valid JSON")
	}
	decoder := json.NewDecoder(bytes.NewReader(data))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(target); err != nil {
		return fmt.Errorf("invalid request: %w", err)
	}
	if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
		return errors.New("invalid request: multiple JSON values")
	}
	return nil
}

func RegisterForm[Values, Response any](name string, validate func(Values) FieldErrors, submit ActionHandler[Values, Response], opts ...ActionOption[Values]) error

{
	if submit == nil {
		return errors.New("host: nil form handler")
	}
	return RegisterAction(name, func(ctx context.Context, session *Session, values Values) (FormResponse[Response], error) {
		if validate != nil {
			if fields := validate(values); len(fields) > 0 {
				return FormResponse[Response]{Fields: fields}, nil
			}
		}
		data, err := submit(ctx, session, values)
		if err != nil {
			return FormResponse[Response]{}, err
		}
		return FormResponse[Response]{Data: data, Valid: true}, nil
	}, opts...)
}

func logLevel() slog.Level

{
	switch strings.ToLower(os.Getenv("RFW_LOG_LEVEL")) {
	case "debug":
		return slog.LevelDebug
	case "warn":
		return slog.LevelWarn
	case "error":
		return slog.LevelError
	default:
		return slog.LevelInfo
	}
}

func NewMuxFS(fsys fs.FS, opts ...MuxOption) *http.ServeMux

{
	runtime := NewWSRuntime(opts...)
	mux := http.NewServeMux()
	if os.Getenv("RFW_DEVTOOLS") != "" {
		mux.Handle("/debug/vars", expvar.Handler())
		mux.HandleFunc("/debug/pprof/", pprof.Index)
		mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
		mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
		mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
		mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
	}
	fileServer := http.FileServerFS(fsys)
	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		if regularFileFS(fsys, r.URL.Path) {
			setWasmEncodingHeaders(w, r.URL.Path, r.URL.Query().Get("v") != "")
			fileServer.ServeHTTP(w, r)
			return
		}
		// Serve index.html only for HTML requests or bare paths, so a missing
		// asset (CSS, JS, image) still returns 404 instead of HTML.
		accept := r.Header.Get("Accept")
		if strings.Contains(accept, "text/html") || r.URL.Path == "/" || r.URL.Path == "" {
			serveIndexFS(w, r, fsys)
			return
		}
		http.NotFound(w, r)
	})
	mux.Handle("/ws", runtime.Guard(websocket.Handler(func(ws *websocket.Conn) {
		wsHandler(ws, runtime)
	})))
	return mux
}

func openWriteTestSocket(t *testing.T) (*websocket.Conn, *websocket.Conn, func())

{
	t.Helper()
	accepted := make(chan *websocket.Conn, 1)
	done := make(chan struct{})
	server := httptest.NewServer(websocket.Handler(func(ws *websocket.Conn) {
		accepted <- ws
		<-done
	}))
	client, err := websocket.Dial("ws"+strings.TrimPrefix(server.URL, "http"), "", server.URL)
	if err != nil {
		server.Close()
		t.Fatalf("dial websocket: %v", err)
	}
	serverSocket := <-accepted
	return client, serverSocket, func() {
		ForgetConnection(serverSocket)
		closeTestResource(t, client)
		closeTestResource(t, serverSocket)
		close(done)
		server.Close()
	}
}

func receiveOrderedMessage(t *testing.T, socket *websocket.Conn) Outbound

{
	t.Helper()
	var raw []byte
	if err := websocket.Message.Receive(socket, &raw); err != nil {
		t.Fatalf("receive websocket message: %v", err)
	}
	var message Outbound
	if err := json.Unmarshal(raw, &message); err != nil {
		t.Fatalf("decode websocket message: %v", err)
	}
	return message
}

func TestSendSessionOutboundSerializesSequenceAndWrite(t *testing.T)

{
	client, server, closeSockets := openWriteTestSocket(t)
	defer closeSockets()
	session := newSession("ordered-write")
	BindSessionConnection(server, session)
	firstEntered := make(chan struct{}, 1)
	firstRelease := make(chan struct{})
	secondEntered := make(chan struct{}, 1)
	firstDone := make(chan struct{})
	secondDone := make(chan struct{})

	go func() {
		SendSessionOutbound(server, session, Outbound{Payload: blockingJSONPayload{
			entered: firstEntered,
			release: firstRelease,
			value:   "first",
		}})
		close(firstDone)
	}()
	<-firstEntered
	go func() {
		SendSessionOutbound(server, session, Outbound{Payload: signalingJSONPayload{
			entered: secondEntered,
			value:   "second",
		}})
		close(secondDone)
	}()

	select {
	case <-secondEntered:
		close(firstRelease)
		<-firstDone
		<-secondDone
		t.Fatal("second message reached the writer before the first completed")
	case <-time.After(50 * time.Millisecond):
	}
	close(firstRelease)
	<-firstDone
	<-secondDone

	first := receiveOrderedMessage(t, client)
	second := receiveOrderedMessage(t, client)
	if first.Sequence != 1 || second.Sequence != 2 {
		t.Fatalf("messages arrived out of order: first=%d second=%d", first.Sequence, second.Sequence)
	}
}

func TestReplaySessionDoesNotInterleaveNewMessages(t *testing.T)

{
	client, server, closeSockets := openWriteTestSocket(t)
	defer closeSockets()
	session := newSession("ordered-replay", sessionOptions{replayLimit: 4})
	BindSessionConnection(server, session)
	firstEntered := make(chan struct{}, 1)
	firstRelease := make(chan struct{})
	newEntered := make(chan struct{}, 1)
	session.PrepareOutbound(Outbound{Payload: blockingJSONPayload{
		entered: firstEntered,
		release: firstRelease,
		value:   "first",
	}})
	session.PrepareOutbound(Outbound{Payload: "second"})
	replayDone := make(chan struct{})
	sendDone := make(chan struct{})

	go func() {
		ReplaySession(server, session, 0)
		close(replayDone)
	}()
	<-firstEntered
	go func() {
		SendSessionOutbound(server, session, Outbound{Payload: signalingJSONPayload{
			entered: newEntered,
			value:   "third",
		}})
		close(sendDone)
	}()

	select {
	case <-newEntered:
		close(firstRelease)
		<-replayDone
		<-sendDone
		t.Fatal("new message reached the writer during replay")
	case <-time.After(50 * time.Millisecond):
	}
	close(firstRelease)
	<-replayDone
	<-sendDone

	for sequence := uint64(1); sequence <= 3; sequence++ {
		if message := receiveOrderedMessage(t, client); message.Sequence != sequence {
			t.Fatalf("unexpected replay order: got %d want %d", message.Sequence, sequence)
		}
	}
}

func TestStaleConnectionDoesNotConsumeSequenceAfterResume(t *testing.T)

{
	_, oldServer, closeOld := openWriteTestSocket(t)
	defer closeOld()
	newClient, newServer, closeNew := openWriteTestSocket(t)
	defer closeNew()
	session := AllocateResumableSession(4)
	defer ReleaseSession(session)
	BindSessionConnection(oldServer, session)
	token := session.ResumeToken()

	SuspendSession(session, time.Second)
	resumed, ok := ResumeSession(token)
	if !ok {
		t.Fatal("session did not resume")
	}
	BindSessionConnection(newServer, resumed)

	staleEntered := make(chan struct{}, 1)
	SendSessionOutbound(oldServer, resumed, Outbound{Payload: signalingJSONPayload{
		entered: staleEntered,
		value:   "stale",
	}})
	select {
	case <-staleEntered:
		t.Fatal("stale connection reached the writer")
	default:
	}

	SendSessionOutbound(newServer, resumed, Outbound{Payload: "current"})
	message := receiveOrderedMessage(t, newClient)
	if message.Sequence != 1 {
		t.Fatalf("stale connection consumed a sequence: got %d want 1", message.Sequence)
	}
}

func TestCustomHandlerDeliveryBindsAcrossResume(t *testing.T)

{
	oldClient, oldServer, closeOld := openWriteTestSocket(t)
	defer closeOld()
	newClient, newServer, closeNew := openWriteTestSocket(t)
	defer closeNew()
	session := AllocateResumableSession(4)
	defer ReleaseSession(session)
	token := session.ResumeToken()

	SendSessionOutbound(oldServer, session, Outbound{Payload: "first"})
	first := receiveOrderedMessage(t, oldClient)
	if first.Sequence != 1 {
		t.Fatalf("first sequence = %d, want 1", first.Sequence)
	}
	SuspendSession(session, time.Second)
	resumed, ok := ResumeSession(token)
	if !ok {
		t.Fatal("session did not resume")
	}

	staleEntered := make(chan struct{}, 1)
	SendSessionOutbound(oldServer, resumed, Outbound{Payload: signalingJSONPayload{
		entered: staleEntered,
		value:   "stale",
	}})
	select {
	case <-staleEntered:
		t.Fatal("stale custom handler connection reached the writer")
	default:
	}

	ReplaySession(newServer, resumed, 0)
	replayed := receiveOrderedMessage(t, newClient)
	if replayed.Sequence != first.Sequence {
		t.Fatalf("replayed sequence = %d, want %d", replayed.Sequence, first.Sequence)
	}

	SendSessionOutbound(oldServer, resumed, Outbound{Payload: signalingJSONPayload{
		entered: staleEntered,
		value:   "stale",
	}})
	select {
	case <-staleEntered:
		t.Fatal("stale custom handler connection reached the writer")
	default:
	}

	SendSessionOutbound(newServer, resumed, Outbound{Payload: "second"})
	second := receiveOrderedMessage(t, newClient)
	if second.Sequence != 2 {
		t.Fatalf("second sequence = %d, want 2", second.Sequence)
	}
}

func TestManagedSessionRejectsAllPriorConnections(t *testing.T)

{
	_, firstServer, closeFirst := openWriteTestSocket(t)
	defer closeFirst()
	_, secondServer, closeSecond := openWriteTestSocket(t)
	defer closeSecond()
	thirdClient, thirdServer, closeThird := openWriteTestSocket(t)
	defer closeThird()
	session := AllocateResumableSession(4)
	defer ReleaseSession(session)
	token := session.ResumeToken()
	BindSessionConnection(firstServer, session)

	SuspendSession(session, time.Second)
	resumed, ok := ResumeSession(token)
	if !ok {
		t.Fatal("first resume failed")
	}
	BindSessionConnection(secondServer, resumed)
	SuspendSession(resumed, time.Second)
	resumed, ok = ResumeSession(token)
	if !ok {
		t.Fatal("second resume failed")
	}
	BindSessionConnection(thirdServer, resumed)

	firstEntered := make(chan struct{}, 1)
	SendSessionOutbound(firstServer, resumed, Outbound{Payload: signalingJSONPayload{
		entered: firstEntered,
		value:   "first-stale",
	}})
	secondEntered := make(chan struct{}, 1)
	SendSessionOutbound(secondServer, resumed, Outbound{Payload: signalingJSONPayload{
		entered: secondEntered,
		value:   "second-stale",
	}})
	select {
	case <-firstEntered:
		t.Fatal("first stale connection reached the writer")
	case <-secondEntered:
		t.Fatal("second stale connection reached the writer")
	default:
	}

	SendSessionOutbound(thirdServer, resumed, Outbound{Payload: "current"})
	message := receiveOrderedMessage(t, thirdClient)
	if message.Sequence != 1 {
		t.Fatalf("stale connection consumed a sequence: got %d want 1", message.Sequence)
	}
}

func DefaultSSCLimits() SSCLimits

{
	return SSCLimits{
		MaxMessageBytes:   1 << 20,
		MaxConnections:    4096,
		MaxSessions:       8192,
		MessagesPerMinute: 600,
		HandlerTimeout:    15 * time.Second,
		ResumeTTL:         2 * time.Minute,
		ReplayMessages:    256,
	}
}

func WithSSCLimits(limits SSCLimits) MuxOption

{
	return func(runtime *WSRuntime) {
		if limits.MaxMessageBytes > 0 {
			runtime.limits.MaxMessageBytes = limits.MaxMessageBytes
		}
		if limits.MaxConnections > 0 {
			runtime.limits.MaxConnections = limits.MaxConnections
		}
		if limits.MaxSessions > 0 {
			runtime.limits.MaxSessions = limits.MaxSessions
		}
		if limits.MessagesPerMinute > 0 {
			runtime.limits.MessagesPerMinute = limits.MessagesPerMinute
		}
		if limits.HandlerTimeout > 0 {
			runtime.limits.HandlerTimeout = limits.HandlerTimeout
		}
		if limits.ResumeTTL > 0 {
			runtime.limits.ResumeTTL = limits.ResumeTTL
		}
		if limits.ReplayMessages > 0 {
			runtime.limits.ReplayMessages = limits.ReplayMessages
		}
	}
}

func TestHostComponent(t *testing.T)

{
	called := false
	hc := NewHostComponent("cmp", func(payload map[string]any) any {
		called = true
		if payload["x"] != 1 {
			t.Fatalf("unexpected payload: %v", payload)
		}
		return "ok"
	})
	Register(hc)
	got, ok := Get("cmp")
	if !ok || got != hc {
		t.Fatalf("component not registered")
	}
	if resp := hc.Handle(map[string]any{"x": 1}); resp != "ok" || !called {
		t.Fatalf("handler not executed or wrong response: %v", resp)
	}
}

func TestHostComponentWithSession(t *testing.T)

{
	hc := NewHostComponentWithSession("withSession", func(session *Session, payload map[string]any) any {
		if session == nil {
			t.Fatalf("session should not be nil")
		}
		store := session.StoreManager().NewStore("test")
		store.Set("value", payload["v"])
		return store.Snapshot()
	})

	sess := newSession("abc")
	resp := hc.HandleWithSession(sess, map[string]any{"v": 42})
	snap, ok := resp.(map[string]any)
	if !ok {
		t.Fatalf("unexpected response type %T", resp)
	}
	if snap["value"] != 42 {
		t.Fatalf("unexpected store snapshot: %v", snap)
	}
	if !hc.SessionAware() {
		t.Fatalf("expected session aware component")
	}
	if hc.StoreManager(sess) != sess.StoreManager() {
		t.Fatalf("StoreManager helper mismatch")
	}
}

func TestLogLevel(t *testing.T)

{
	t.Setenv("RFW_LOG_LEVEL", "debug")
	if lvl := logLevel(); lvl.String() != "DEBUG" {
		t.Fatalf("expected DEBUG level, got %s", lvl)
	}
	t.Setenv("RFW_LOG_LEVEL", "warn")
	if lvl := logLevel(); lvl.String() != "WARN" {
		t.Fatalf("expected WARN level, got %s", lvl)
	}
	t.Setenv("RFW_LOG_LEVEL", "")
	if lvl := logLevel(); lvl.String() != "INFO" {
		t.Fatalf("expected INFO level, got %s", lvl)
	}
}

func TestSessionIsolation(t *testing.T)

{
	t.Helper()

	registry = make(map[string]*ServerComponent)

	const componentName = "SessionHost"
	Register(NewHostComponentWithSession(componentName, func(session *Session, payload map[string]any) any {
		const storeKey = "counter"
		storeVal, ok := session.ContextGet(storeKey)
		var store *state.Store
		if ok {
			store = storeVal.(*state.Store)
		} else {
			store = session.StoreManager().NewStore("counter")
			store.Set("value", 0)
			session.ContextSet(storeKey, store)
		}
		if inc, ok := payload["increment"].(bool); ok && inc {
			current, _ := store.Get("value").(int)
			store.Set("value", current+1)
		}
		return map[string]any{"value": store.Get("value")}
	}))

	root := t.TempDir()
	// Ensure an index exists so NewMux can serve fallback responses without error.
	if err := os.WriteFile(filepath.Join(root, "index.html"), []byte("ok"), 0o600); err != nil {
		t.Fatalf("write index: %v", err)
	}

	srv := httptest.NewServer(loggingMiddleware(NewMux(root)))
	defer srv.Close()

	wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "/ws"

	type sessionConn struct {
		ws  *websocket.Conn
		id  string
		idx int
	}

	dial := func(idx int) sessionConn {
		ws, err := websocket.Dial(wsURL, "", srv.URL)
		if err != nil {
			t.Fatalf("dial %d: %v", idx, err)
		}
		init := map[string]any{
			"component": componentName,
			"payload":   map[string]any{"init": true},
		}
		raw, err := json.Marshal(init)
		if err != nil {
			t.Fatalf("marshal init %d: %v", idx, err)
		}
		if err := websocket.Message.Send(ws, raw); err != nil {
			t.Fatalf("send init %d: %v", idx, err)
		}
		var respRaw []byte
		if err := websocket.Message.Receive(ws, &respRaw); err != nil {
			t.Fatalf("recv init %d: %v", idx, err)
		}
		var resp struct {
			Component string         `json:"component"`
			Payload   map[string]any `json:"payload"`
			Session   string         `json:"session"`
		}
		if err := json.Unmarshal(respRaw, &resp); err != nil {
			t.Fatalf("unmarshal init %d: %v", idx, err)
		}
		if resp.Session == "" {
			t.Fatalf("session id missing for conn %d", idx)
		}
		if resp.Component != componentName {
			t.Fatalf("unexpected component %s", resp.Component)
		}
		if val, ok := resp.Payload["value"].(float64); !ok || val != 0 {
			t.Fatalf("unexpected init value for conn %d: %v", idx, resp.Payload)
		}
		return sessionConn{ws: ws, id: resp.Session, idx: idx}
	}

	sessions := []sessionConn{dial(0), dial(1)}
	defer func() {
		for _, sc := range sessions {
			closeTestResource(t, sc.ws)
		}
	}()

	counts := []int{5, 2}
	if len(counts) != len(sessions) {
		t.Fatalf("mismatched counts")
	}

	errCh := make(chan error, len(sessions))
	var wg sync.WaitGroup
	for i, sc := range sessions {
		wg.Add(1)
		count := counts[i]
		go func(sc sessionConn, target int) {
			defer wg.Done()
			for j := 0; j < target; j++ {
				payload := map[string]any{
					"component": componentName,
					"payload":   map[string]any{"increment": true},
				}
				raw, err := json.Marshal(payload)
				if err != nil {
					errCh <- fmt.Errorf("marshal increment idx=%d: %w", sc.idx, err)
					return
				}
				if err := websocket.Message.Send(sc.ws, raw); err != nil {
					errCh <- fmt.Errorf("send increment idx=%d: %w", sc.idx, err)
					return
				}
				var respRaw []byte
				if err := websocket.Message.Receive(sc.ws, &respRaw); err != nil {
					errCh <- fmt.Errorf("recv increment idx=%d: %w", sc.idx, err)
					return
				}
				var resp struct {
					Component string         `json:"component"`
					Payload   map[string]any `json:"payload"`
					Session   string         `json:"session"`
				}
				if err := json.Unmarshal(respRaw, &resp); err != nil {
					errCh <- fmt.Errorf("unmarshal increment idx=%d: %w", sc.idx, err)
					return
				}
				if resp.Session != sc.id {
					errCh <- fmt.Errorf("response session mismatch idx=%d: got %s want %s", sc.idx, resp.Session, sc.id)
					return
				}
			}
			errCh <- nil
		}(sc, count)
	}

	wg.Wait()
	close(errCh)
	for err := range errCh {
		if err != nil {
			t.Fatal(err)
		}
	}

	for i, sc := range sessions {
		sess, ok := SessionByID(sc.id)
		if !ok {
			t.Fatalf("session %d not found", i)
		}
		snap := sess.Snapshot()
		module := snap["default"]
		if module == nil {
			t.Fatalf("session %d missing default module snapshot", i)
		}
		counter := module["counter"]
		if counter == nil {
			t.Fatalf("session %d missing counter store", i)
		}
		val, ok := counter["value"].(int)
		if !ok {
			t.Fatalf("session %d missing value entry: %v", i, counter)
		}
		if val != counts[i] {
			t.Fatalf("session %d got value %d want %d", i, val, counts[i])
		}
	}

	if sessions[0].id == sessions[1].id {
		t.Fatal("session ids should differ")
	}
}

func openProtocolSocket(t *testing.T, opts ...MuxOption) (*websocket.Conn, func())

{
	t.Helper()
	server := httptest.NewServer(NewMux(t.TempDir(), opts...))
	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/ws"
	socket, err := websocket.Dial(wsURL, "", server.URL)
	if err != nil {
		server.Close()
		t.Fatalf("dial websocket: %v", err)
	}
	return socket, func() {
		closeTestResource(t, socket)
		server.Close()
	}
}

func receiveProtocolMessage(t *testing.T, socket *websocket.Conn) Outbound

{
	t.Helper()
	var data []byte
	if err := websocket.Message.Receive(socket, &data); err != nil {
		t.Fatalf("receive message: %v", err)
	}
	var message Outbound
	if err := json.Unmarshal(data, &message); err != nil {
		t.Fatalf("decode message: %v", err)
	}
	return message
}

func TestWSTypedActionRejectsUnknownFields(t *testing.T)

{
	type request struct {
		Value int `json:"value"`
	}
	const action = "test.ws.strict"
	if err := RegisterAction(action, func(_ context.Context, _ *Session, request request) (request, error) {
		return request, nil
	}); err != nil {
		t.Fatalf("register action: %v", err)
	}
	socket, closeSocket := openProtocolSocket(t)
	defer closeSocket()

	sendProtocolMessage(t, socket, Inbound{
		Action:   action,
		ID:       "strict",
		Sequence: 1,
		Payload:  map[string]any{"value": 1, "admin": true},
	})
	response := receiveProtocolMessage(t, socket)
	if response.ID != "strict" || response.Error == nil || response.Error.Code != "invalid_request" {
		t.Fatalf("unexpected strict response: %#v", response)
	}
}

func TestWSMessageAuthorizationRunsBeforeAction(t *testing.T)

{
	type request struct {
		Value int `json:"value"`
	}
	const action = "test.ws.authorized"
	called := false
	if err := RegisterAction(action, func(_ context.Context, _ *Session, request request) (request, error) {
		called = true
		return request, nil
	}); err != nil {
		t.Fatalf("register action: %v", err)
	}
	socket, closeSocket := openProtocolSocket(t, WithSSCAuthorizer(func(_ context.Context, _ *Session, message Inbound) error {
		if message.Action == action {
			return errors.New("denied")
		}
		return nil
	}))
	defer closeSocket()

	sendProtocolMessage(t, socket, Inbound{Action: action, ID: "denied", Sequence: 1})
	response := receiveProtocolMessage(t, socket)
	if response.Error == nil || response.Error.Code != "forbidden" || called {
		t.Fatalf("authorization failed closed incorrectly: response=%#v called=%v", response, called)
	}
}

func TestWSRateLimitRejectsExcessMessages(t *testing.T)

{
	type request struct{}
	const action = "test.ws.rate"
	if err := RegisterAction(action, func(_ context.Context, _ *Session, _ request) (string, error) {
		return "ok", nil
	}); err != nil {
		t.Fatalf("register action: %v", err)
	}
	socket, closeSocket := openProtocolSocket(t, WithSSCLimits(SSCLimits{MessagesPerMinute: 1}))
	defer closeSocket()

	sendProtocolMessage(t, socket, Inbound{Action: action, ID: "first", Sequence: 1})
	if response := receiveProtocolMessage(t, socket); response.Error != nil {
		t.Fatalf("first message rejected: %#v", response)
	}
	sendProtocolMessage(t, socket, Inbound{Action: action, ID: "second", Sequence: 2})
	response := receiveProtocolMessage(t, socket)
	if response.Error == nil || response.Error.Code != "rate_limited" {
		t.Fatalf("rate limit did not reject second message: %#v", response)
	}
	if response.Ack != 2 {
		t.Fatalf("rate-limited message was not acknowledged: %#v", response)
	}
}

func TestWSActionTimeout(t *testing.T)

{
	type request struct{}
	const action = "test.ws.timeout"
	if err := RegisterAction(action, func(_ context.Context, _ *Session, _ request) (string, error) {
		time.Sleep(50 * time.Millisecond)
		return "late", nil
	}); err != nil {
		t.Fatalf("register action: %v", err)
	}
	socket, closeSocket := openProtocolSocket(t, WithSSCLimits(SSCLimits{HandlerTimeout: 5 * time.Millisecond}))
	defer closeSocket()

	sendProtocolMessage(t, socket, Inbound{Action: action, ID: "timeout", Sequence: 1})
	response := receiveProtocolMessage(t, socket)
	if response.Error == nil || response.Error.Code != "action_timeout" {
		t.Fatalf("slow action did not time out: %#v", response)
	}
}

func TestWSActionPanicReturnsPublicError(t *testing.T)

{
	type request struct{}
	const action = "test.ws.panic"
	if err := RegisterAction(action, func(_ context.Context, _ *Session, _ request) (string, error) {
		panic("private handler detail")
	}); err != nil {
		t.Fatalf("register action: %v", err)
	}
	socket, closeSocket := openProtocolSocket(t)
	defer closeSocket()

	sendProtocolMessage(t, socket, Inbound{Action: action, ID: "panic", Sequence: 1})
	response := receiveProtocolMessage(t, socket)
	if response.Error == nil || response.Error.Code != "action_failed" || response.Error.Message != "action failed" {
		t.Fatalf("panic details crossed the protocol: %#v", response)
	}
}

func TestWSRejectsOversizedFrame(t *testing.T)

{
	socket, closeSocket := openProtocolSocket(t, WithSSCLimits(SSCLimits{MaxMessageBytes: 128}))
	defer closeSocket()

	data, err := json.Marshal(Inbound{
		Component: "oversized",
		Sequence:  1,
		Payload:   map[string]any{"value": strings.Repeat("x", 512)},
	})
	if err != nil {
		t.Fatalf("marshal oversized message: %v", err)
	}
	if err := websocket.Message.Send(socket, data); err != nil {
		t.Fatalf("send oversized message: %v", err)
	}
	var response []byte
	if err := websocket.Message.Receive(socket, &response); err == nil {
		t.Fatalf("oversized frame was accepted: %s", response)
	}
}

func TestWSSessionResumeReplaysUnacknowledgedResponse(t *testing.T)

{
	type request struct{}
	type response struct {
		Count int `json:"count"`
	}
	const action = "test.ws.resume"
	if err := RegisterAction(action, func(_ context.Context, session *Session, _ request) (response, error) {
		count := 0
		if stored, ok := session.ContextGet("count"); ok {
			count = stored.(int)
		}
		count++
		session.ContextSet("count", count)
		return response{Count: count}, nil
	}); err != nil {
		t.Fatalf("register action: %v", err)
	}

	sessionMu.RLock()
	maxSessions := len(sessions) + 1
	sessionMu.RUnlock()
	server := httptest.NewServer(NewMux(t.TempDir(), WithSSCLimits(SSCLimits{
		ResumeTTL:      time.Second,
		ReplayMessages: 8,
		MaxSessions:    maxSessions,
	})))
	defer server.Close()
	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/ws"
	dial := func() *websocket.Conn {
		socket, err := websocket.Dial(wsURL, "", server.URL)
		if err != nil {
			t.Fatalf("dial websocket: %v", err)
		}
		return socket
	}

	firstSocket := dial()
	sendProtocolMessage(t, firstSocket, Inbound{Action: action, ID: "one", Sequence: 1})
	first := receiveProtocolMessage(t, firstSocket)
	sendProtocolMessage(t, firstSocket, Inbound{Action: action, ID: "two", Sequence: 2, Ack: first.Sequence})
	second := receiveProtocolMessage(t, firstSocket)
	token := second.ResumeToken
	sessionID := second.Session
	closeTestResource(t, firstSocket)

	var (
		secondSocket *websocket.Conn
		replayed     Outbound
		current      Outbound
	)
	deadline := time.Now().Add(time.Second)
	for {
		secondSocket = dial()
		sendProtocolMessage(t, secondSocket, Inbound{
			Action:      action,
			ID:          "three",
			Sequence:    3,
			Ack:         first.Sequence,
			ResumeToken: token,
		})
		replayed = receiveProtocolMessage(t, secondSocket)
		if replayed.Session == sessionID {
			current = receiveProtocolMessage(t, secondSocket)
			break
		}
		closeTestResource(t, secondSocket)
		if time.Now().After(deadline) {
			t.Fatalf("session did not become resumable: %#v", replayed)
		}
		time.Sleep(10 * time.Millisecond)
	}
	defer closeTestResource(t, secondSocket)
	if replayed.Sequence != second.Sequence || replayed.ID != "two" {
		t.Fatalf("unexpected replay: %#v", replayed)
	}
	if current.Session != sessionID || current.ID != "three" {
		t.Fatalf("session did not resume: %#v", current)
	}
	payload, ok := current.Payload.(map[string]any)
	if !ok || payload["count"] != float64(3) {
		t.Fatalf("session state was not retained: %#v", current.Payload)
	}
	if session, ok := SessionByID(sessionID); ok {
		ReleaseSession(session)
	}
}

func TestWSSessionResumeExpires(t *testing.T)

{
	type request struct{}
	const action = "test.ws.expiry"
	if err := RegisterAction(action, func(_ context.Context, _ *Session, _ request) (string, error) {
		return "ok", nil
	}); err != nil {
		t.Fatalf("register action: %v", err)
	}
	server := httptest.NewServer(NewMux(t.TempDir(), WithSSCLimits(SSCLimits{ResumeTTL: 10 * time.Millisecond})))
	defer server.Close()
	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/ws"

	first, err := websocket.Dial(wsURL, "", server.URL)
	if err != nil {
		t.Fatalf("dial first socket: %v", err)
	}
	sendProtocolMessage(t, first, Inbound{Action: action, ID: "first", Sequence: 1})
	response := receiveProtocolMessage(t, first)
	token := response.ResumeToken
	closeTestResource(t, first)
	time.Sleep(40 * time.Millisecond)

	second, err := websocket.Dial(wsURL, "", server.URL)
	if err != nil {
		t.Fatalf("dial second socket: %v", err)
	}
	defer closeTestResource(t, second)
	sendProtocolMessage(t, second, Inbound{
		Action:      action,
		ID:          "second",
		Sequence:    2,
		ResumeToken: token,
	})
	rejected := receiveProtocolMessage(t, second)
	if rejected.Control != "resume_rejected" || rejected.Error == nil {
		t.Fatalf("expired session resumed: %#v", rejected)
	}
}

func wsHandler(ws *websocket.Conn, runtime *WSRuntime)

{
	if !runtime.AcquireConnection() {
		SendOutbound(ws, Outbound{Error: NewActionError("connection_limit", "connection limit reached")})
		if err := ws.Close(); err != nil {
			log.Printf("close rejected websocket: %v", err)
		}
		return
	}
	defer runtime.ReleaseConnection()
	runtime.ConfigureConnection(ws)

	var session *Session
	var subscribed []string
	defer func() {
		connMu.Lock()
		for _, name := range subscribed {
			if set, ok := connections[name]; ok {
				delete(set, ws)
				if len(set) == 0 {
					delete(connections, name)
				}
			}
		}
		connMu.Unlock()
		SuspendSession(session, runtime.ResumeTTL())
		ForgetConnection(ws)
		if err := ws.Close(); err != nil {
			log.Printf("close websocket: %v", err)
		}
	}()
	for {
		var raw []byte
		if err := websocket.Message.Receive(ws, &raw); err != nil {
			if err == io.EOF {
				break
			}
			log.Printf("recv: %v", err)
			return
		}
		var msg Inbound
		if err := json.Unmarshal(raw, &msg); err != nil {
			log.Printf("unmarshal: %v", err)
			continue
		}
		if session == nil {
			var resumed bool
			var err error
			session, resumed, err = runtime.OpenSession(ws.Request(), msg.ResumeToken)
			if err != nil {
				SendOutbound(ws, Outbound{Error: NewActionError("session_rejected", "session rejected")})
				return
			}
			BindSessionConnection(ws, session)
			if resumed {
				ReplaySession(ws, session, msg.Ack)
			} else if msg.ResumeToken != "" {
				SendSessionOutbound(ws, session, Outbound{
					Control: "resume_rejected",
					Error:   NewActionError("resume_rejected", "session could not be resumed"),
				})
			}
		}
		session.Acknowledge(msg.Ack)
		if err := session.AcceptInbound(msg.Sequence); err != nil {
			if errors.Is(err, ErrDuplicateMessage) {
				continue
			}
			SendSessionOutbound(ws, session, Outbound{
				ID:     msg.ID,
				Action: msg.Action,
				Error:  NewActionError("sequence_gap", "client message sequence gap"),
			})
			continue
		}
		if !session.AllowMessage(runtime.MessagesPerMinute()) {
			SendSessionOutbound(ws, session, Outbound{
				ID:     msg.ID,
				Action: msg.Action,
				Error:  NewActionError("rate_limited", "message rate limit exceeded"),
			})
			continue
		}
		authorizeCtx, cancelAuthorize := runtime.HandlerContext(context.Background())
		authorizeErr := runtime.Authorize(authorizeCtx, session, msg)
		cancelAuthorize()
		if authorizeErr != nil {
			SendSessionOutbound(ws, session, Outbound{
				Component: msg.Component,
				Action:    msg.Action,
				ID:        msg.ID,
				Error:     NewActionError("forbidden", "message forbidden"),
			})
			continue
		}
		if msg.Action != "" {
			payload, actionErr := runtime.DispatchAction(context.Background(), session, msg)
			SendSessionOutbound(ws, session, Outbound{
				Action:  msg.Action,
				ID:      msg.ID,
				Payload: payload,
				Error:   actionErr,
			})
			continue
		}
		if msg.Component != "" && msg.Payload != nil && msg.Payload["unsubscribe"] == true {
			connMu.Lock()
			if set, ok := connections[msg.Component]; ok {
				delete(set, ws)
				if len(set) == 0 {
					delete(connections, msg.Component)
				}
			}
			for index, name := range subscribed {
				if name == msg.Component {
					subscribed = append(subscribed[:index], subscribed[index+1:]...)
					break
				}
			}
			connMu.Unlock()
			SendSessionOutbound(ws, session, Outbound{Component: msg.Component, Control: "unsubscribed"})
			continue
		}
		if hc, ok := Get(msg.Component); ok {
			connMu.Lock()
			if _, ok := connections[msg.Component]; !ok {
				connections[msg.Component] = make(map[*websocket.Conn]*Session)
			}
			if _, tracked := connections[msg.Component][ws]; !tracked {
				connections[msg.Component][ws] = session
				subscribed = append(subscribed, msg.Component)
			}
			connMu.Unlock()
			resp := hc.HandleWithSession(session, msg.Payload)
			if resp != nil {
				switch v := resp.(type) {
				case *InitSnapshot:
					if v != nil {
						SendSessionOutbound(ws, session, Outbound{Component: msg.Component, ID: msg.ID, Payload: map[string]any{"initSnapshot": v}})
					}
					continue
				case InitSnapshot:
					SendSessionOutbound(ws, session, Outbound{Component: msg.Component, ID: msg.ID, Payload: map[string]any{"initSnapshot": v}})
					continue
				default:
					SendSessionOutbound(ws, session, Outbound{Component: msg.Component, ID: msg.ID, Payload: resp})
					continue
				}
			}
			if msg.Payload != nil && msg.Payload["init"] == true {
				SendSessionOutbound(ws, session, Outbound{
					Component: msg.Component,
					Payload:   map[string]any{"session": session.ID()},
				})
				continue
			}
		}
		SendSessionOutbound(ws, session, Outbound{Control: "ack"})
	}
}

func Broadcast(name string, payload any, opts ...BroadcastOption)

{
	var options BroadcastOptions
	for _, opt := range opts {
		opt(&options)
	}

	// Snapshot the (conn, session) pairs under the lock: wsHandler mutates the
	// connection map on subscribe/disconnect, so iterating it after releasing
	// connMu races with those writes. Sends happen outside the lock.
	type target struct {
		ws      *websocket.Conn
		session *Session
	}
	connMu.RLock()
	targets := make([]target, 0, len(connections[name]))
	for ws, session := range connections[name] {
		targets = append(targets, target{ws: ws, session: session})
	}
	connMu.RUnlock()

	for _, t := range targets {
		if options.Session != "" && t.session.ID() != options.Session {
			continue
		}
		SendSessionOutbound(t.ws, t.session, Outbound{Component: name, Payload: payload})
	}
}

func ReplaySession(ws *websocket.Conn, session *Session, acknowledged uint64)

{
	if session == nil {
		return
	}
	session.outboundMu.Lock()
	defer session.outboundMu.Unlock()
	accepted, _ := sessionAcceptsConnection(session, ws, true)
	if !accepted {
		return
	}
	lock := connectionWriteLock(ws)
	lock.Lock()
	defer lock.Unlock()
	messages, err := session.ReplayAfter(acknowledged)
	if err != nil {
		sendOutboundUnlocked(ws, session.PrepareOutbound(Outbound{
			Error: NewActionError("resync_required", "message history is no longer available"),
		}))
		return
	}
	for _, message := range messages {
		sendOutboundUnlocked(ws, message)
	}
}

func SendSessionOutbound(ws *websocket.Conn, session *Session, out Outbound)

{
	if session == nil {
		return
	}
	session.outboundMu.Lock()
	defer session.outboundMu.Unlock()
	accepted, handoffPending := sessionAcceptsConnection(session, ws, false)
	if !accepted {
		if handoffPending {
			logger.Debug("session outbound dropped before connection handoff", "session", session.ID())
		}
		return
	}
	lock := connectionWriteLock(ws)
	lock.Lock()
	defer lock.Unlock()
	sendOutboundUnlocked(ws, session.PrepareOutbound(out))
}

func BindSessionConnection(ws *websocket.Conn, session *Session)

{
	if session == nil {
		return
	}
	session.outboundMu.Lock()
	accepted, _ := sessionAcceptsConnection(session, ws, true)
	if accepted {
		session.connectionManaged = true
	}
	session.outboundMu.Unlock()
}

func sessionAcceptsConnection(session *Session, ws *websocket.Conn, handoff bool) (bool, bool)

{
	if ws == nil {
		return false, false
	}
	session.deliveryMu.Lock()
	active := session.attached && !session.released
	resumePending := session.resumePending
	if active && resumePending && handoff {
		session.resumePending = false
	}
	session.deliveryMu.Unlock()
	if !active {
		return false, false
	}
	if resumePending && !handoff {
		return false, true
	}
	if session.connection == ws {
		return true, false
	}
	if session.connection != nil || (session.connectionManaged && !handoff) {
		return false, false
	}
	session.connection = ws
	return true, false
}