host

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 detachedSession(t *testing.T, runtime *WSRuntime, user string) *Session

{
	t.Helper()
	session, resumed, err := runtime.OpenSession(resumeRequest(t, user), "")
	if err != nil {
		t.Fatalf("open session: %v", err)
	}
	if resumed || session.ResumeToken() == "" {
		t.Fatalf("expected a new resumable session, resumed=%v token=%q", resumed, session.ResumeToken())
	}
	SuspendSession(session, runtime.ResumeTTL())
	return session
}

func assertDetached(t *testing.T, session *Session)

{
	t.Helper()
	session.deliveryMu.Lock()
	attached := session.attached
	expires := session.expires
	session.deliveryMu.Unlock()
	if attached {
		t.Fatal("rejected attempt attached the retained session")
	}
	if expires.IsZero() {
		t.Fatal("rejected attempt cleared the retention deadline")
	}
}

func TestResumeAuthorizerRejectsAnotherUser(t *testing.T)

{
	initializations := 0
	runtime := resumableRuntime(
		WithSSCSessionInitializer(func(request *http.Request, session *Session) error {
			initializations++
			return bindResumeIdentity(request, session)
		}),
		WithSSCResumeAuthorizer(authorizeSameUserResume),
	)
	owner := detachedSession(t, runtime, "alice")
	defer ReleaseSession(owner)
	token := owner.ResumeToken()

	intruder, resumed, err := runtime.OpenSession(resumeRequest(t, "mallory"), token)
	if err != nil {
		t.Fatalf("open session for rejected resume: %v", err)
	}
	defer ReleaseSession(intruder)
	if resumed || intruder == owner {
		t.Fatalf("foreign request resumed the retained session: resumed=%v", resumed)
	}
	if user, _ := owner.ContextGet("user"); user != "alice" {
		t.Fatalf("retained session identity changed to %v", user)
	}
	assertDetached(t, owner)

	restored, resumed, err := runtime.OpenSession(resumeRequest(t, "alice"), token)
	if err != nil {
		t.Fatalf("open session for owner resume: %v", err)
	}
	if !resumed || restored != owner {
		t.Fatalf("owner could not resume after a rejected attempt: resumed=%v", resumed)
	}
	if user, _ := restored.ContextGet("user"); user != "alice" {
		t.Fatalf("resume rebound the session identity to %v", user)
	}
	// The two new sessions were initialized; the resume was not.
	if initializations != 2 {
		t.Fatalf("session initializations = %d, want 2", initializations)
	}
}

func TestResumeAuthorizerRejectsRevokedIdentity(t *testing.T)

{
	revoked := map[string]bool{"alice": true}
	runtime := resumableRuntime(
		WithSSCSessionInitializer(bindResumeIdentity),
		WithSSCResumeAuthorizer(func(request *http.Request, session *Session) error {
			if err := authorizeSameUserResume(request, session); err != nil {
				return err
			}
			if revoked[request.Header.Get(resumeIdentityHeader)] {
				return errors.New("credential revoked")
			}
			return nil
		}),
	)
	owner := detachedSession(t, runtime, "alice")
	defer ReleaseSession(owner)
	token := owner.ResumeToken()

	replacement, resumed, err := runtime.OpenSession(resumeRequest(t, "alice"), token)
	if err != nil {
		t.Fatalf("open session for revoked resume: %v", err)
	}
	defer ReleaseSession(replacement)
	if resumed || replacement == owner {
		t.Fatalf("revoked identity resumed the session: resumed=%v", resumed)
	}
	assertDetached(t, owner)

	revoked["alice"] = false
	restored, resumed, err := runtime.OpenSession(resumeRequest(t, "alice"), token)
	if err != nil {
		t.Fatalf("open session after revocation cleared: %v", err)
	}
	if !resumed || restored != owner {
		t.Fatalf("cleared credential could not resume: resumed=%v", resumed)
	}
}

func TestConcurrentResumeAttachesOneAuthorizedSession(t *testing.T)

{
	authorizing := make(chan struct{}, 2)
	release := make(chan struct{})
	runtime := resumableRuntime(
		WithSSCSessionInitializer(bindResumeIdentity),
		WithSSCResumeAuthorizer(func(request *http.Request, session *Session) error {
			if err := authorizeSameUserResume(request, session); err != nil {
				return err
			}
			authorizing <- struct{}{}
			<-release
			return nil
		}),
	)
	owner := detachedSession(t, runtime, "alice")
	defer ReleaseSession(owner)
	token := owner.ResumeToken()

	type outcome struct {
		session *Session
		resumed bool
		err     error
	}
	requests := []*http.Request{resumeRequest(t, "alice"), resumeRequest(t, "alice")}
	results := make(chan outcome, len(requests))
	for _, request := range requests {
		go func(request *http.Request) {
			session, resumed, err := runtime.OpenSession(request, token)
			results <- outcome{session: session, resumed: resumed, err: err}
		}(request)
	}
	// Both attempts hold an authorized candidate before either commits.
	<-authorizing
	<-authorizing
	close(release)

	attached := 0
	for range requests {
		result := <-results
		if result.err != nil {
			t.Fatalf("concurrent open session: %v", result.err)
		}
		if result.resumed {
			attached++
			if result.session != owner {
				t.Fatal("resume attached a session the request was not authorized for")
			}
			continue
		}
		if result.session == owner {
			t.Fatal("losing attempt received the retained session")
		}
		ReleaseSession(result.session)
	}
	if attached != 1 {
		t.Fatalf("resumed sessions = %d, want 1", attached)
	}
}

func TestResumeWithoutAuthorizerKeepsTokenBehavior(t *testing.T)

{
	runtime := resumableRuntime()
	owner := detachedSession(t, runtime, "alice")
	defer ReleaseSession(owner)

	restored, resumed, err := runtime.OpenSession(resumeRequest(t, "mallory"), owner.ResumeToken())
	if err != nil {
		t.Fatalf("open session: %v", err)
	}
	if !resumed || restored != owner {
		t.Fatalf("token resume changed without an authorizer: resumed=%v", resumed)
	}
}

func TestReleasedSessionTokenStaysUnresumable(t *testing.T)

{
	handed := 0
	runtime := resumableRuntime(
		WithSSCSessionInitializer(bindResumeIdentity),
		WithSSCResumeAuthorizer(func(request *http.Request, session *Session) error {
			handed++
			return authorizeSameUserResume(request, session)
		}),
	)
	owner := detachedSession(t, runtime, "alice")
	token := owner.ResumeToken()

	ReleaseSession(owner)
	ReleaseSession(owner)

	for attempt := 0; attempt < 2; attempt++ {
		session, resumed, err := runtime.OpenSession(resumeRequest(t, "alice"), token)
		if err != nil {
			t.Fatalf("attempt %d: open session: %v", attempt, err)
		}
		t.Cleanup(func() { ReleaseSession(session) })
		if resumed || session == owner {
			t.Fatalf("attempt %d: the released session was resumed: resumed=%v", attempt, resumed)
		}
		if session.ResumeToken() == "" || session.ResumeToken() == token {
			t.Fatalf("attempt %d: the fresh session carries token %q", attempt, session.ResumeToken())
		}
	}
	if handed != 0 {
		t.Fatalf("the authorizer was handed a released session %d times", handed)
	}
	if _, registered := SessionByID(owner.ID()); registered {
		t.Fatal("the released session is still registered")
	}
	// Detaching it again must not put its token back into circulation.
	SuspendSession(owner, runtime.ResumeTTL())
	if resumed, ok := ResumeSession(token); ok {
		t.Fatalf("the released token reattached session %q", resumed.ID())
	}
}

func TestResumeAuthorizerSkippedForUnknownToken(t *testing.T)

{
	calls := 0
	runtime := resumableRuntime(
		WithSSCSessionInitializer(bindResumeIdentity),
		WithSSCResumeAuthorizer(func(*http.Request, *Session) error {
			calls++
			return nil
		}),
	)
	session, resumed, err := runtime.OpenSession(resumeRequest(t, "alice"), "unknown-token")
	if err != nil {
		t.Fatalf("open session: %v", err)
	}
	defer ReleaseSession(session)
	if resumed || calls != 0 {
		t.Fatalf("unknown token reached the authorizer: resumed=%v calls=%d", resumed, calls)
	}
}

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 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 TestResolveTransportPrecedence(t *testing.T)

{
	if runtime.GOOS == "js" {
		t.Skip("os.Chdir is not implemented on js")
	}
	dir := t.TempDir()
	previous, err := os.Getwd()
	if err != nil {
		t.Fatal(err)
	}
	if err := os.Chdir(dir); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = os.Chdir(previous) })

	if got := ResolveTransport(); got != TransportWebSocket {
		t.Fatalf("default = %q", got)
	}
	if err := os.WriteFile(filepath.Join(dir, "rfw.json"), []byte(`{"transport":"streambus"}`), 0o600); err != nil {
		t.Fatal(err)
	}
	if got := ResolveTransport(); got != TransportStreamBus {
		t.Fatalf("manifest = %q", got)
	}
	t.Setenv("RFW_TRANSPORT", "ws")
	if got := ResolveTransport(); got != TransportWebSocket {
		t.Fatalf("environment = %q", got)
	}
}

func TestResolveTransportAliases(t *testing.T)

{
	for value, want := range map[string]Transport{
		"websocket":      TransportWebSocket,
		"WS":             TransportWebSocket,
		"streambus":      TransportStreamBus,
		"webtransport":   TransportStreamBus,
		"warp-streambus": TransportStreamBus,
		"auto":           TransportAuto,
	} {
		if got := normalizeTransport(value); got != want {
			t.Errorf("normalizeTransport(%q) = %q, want %q", value, got, want)
		}
	}
}

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 TestFrameRoundTrip(t *testing.T)

{
	var stream bytes.Buffer
	want := []byte(`{"component":"Counter","payload":{"value":42}}`)
	if err := writeFrame(&stream, want); err != nil {
		t.Fatal(err)
	}
	got, err := readFrame(bufio.NewReader(&stream), 1024)
	if err != nil {
		t.Fatal(err)
	}
	if !bytes.Equal(got, want) {
		t.Fatalf("frame = %q, want %q", got, want)
	}
}

func buildDir(t *testing.T, artifacts map[string]string) (*http.ServeMux, map[string]int)

{
	t.Helper()
	dir := t.TempDir()
	client := filepath.Join(dir, "client")
	if err := os.MkdirAll(client, 0o750); err != nil {
		t.Fatalf("mkdir client: %v", err)
	}
	sizes := map[string]int{}
	for name, body := range artifacts {
		if err := os.WriteFile(filepath.Join(client, name), []byte(body), 0o600); err != nil {
			t.Fatalf("write %s: %v", name, err)
		}
		sizes[name] = len(body)
	}
	if err := os.WriteFile(filepath.Join(client, "index.html"), []byte("<html></html>"), 0o600); err != nil {
		t.Fatalf("write index: %v", err)
	}
	return NewMux(client), sizes
}

func TestNegotiationPrefersBrotli(t *testing.T)

{
	mux, sizes := buildDir(t, allArtifacts)
	resp := request(t, mux, "/app.wasm?v=abc123", "gzip, deflate, br")
	defer func() { _ = resp.Body.Close() }()

	if got := resp.Header.Get("Content-Encoding"); got != "br" {
		t.Fatalf("Content-Encoding = %q, want br", got)
	}
	if got := resp.Header.Get("Content-Type"); got != "application/wasm" {
		t.Fatalf("Content-Type = %q", got)
	}
	if got := resp.Header.Get("Content-Length"); got != strconv.Itoa(sizes["app.wasm.br"]) {
		t.Fatalf("Content-Length = %q, want the compressed size %d", got, sizes["app.wasm.br"])
	}
	if got := resp.Header.Get("Vary"); got != "Accept-Encoding" {
		t.Fatalf("Vary = %q", got)
	}
	if got := resp.Header.Get("Cache-Control"); got != "public, max-age=31536000, immutable" {
		t.Fatalf("Cache-Control on a versioned URL = %q", got)
	}
}

func TestNegotiationFallsBackToGzip(t *testing.T)

{
	mux, sizes := buildDir(t, allArtifacts)
	resp := request(t, mux, "/app.wasm", "gzip, deflate")
	defer func() { _ = resp.Body.Close() }()

	if got := resp.Header.Get("Content-Encoding"); got != "gzip" {
		t.Fatalf("Content-Encoding = %q, want gzip", got)
	}
	if got := resp.Header.Get("Content-Length"); got != strconv.Itoa(sizes["app.wasm.gz"]) {
		t.Fatalf("Content-Length = %q, want %d", got, sizes["app.wasm.gz"])
	}
	// An unversioned URL must revalidate or a release cannot replace it.
	if got := resp.Header.Get("Cache-Control"); got != "no-cache" {
		t.Fatalf("Cache-Control on an unversioned URL = %q", got)
	}
}

func TestNegotiationNeverSendsUnrequestedBrotli(t *testing.T)

{
	mux, _ := buildDir(t, allArtifacts)
	for _, accept := range []string{"gzip", "gzip, deflate", "identity", "br;q=0, gzip"} {
		resp := request(t, mux, "/app.wasm", accept)
		encoding := resp.Header.Get("Content-Encoding")
		_ = resp.Body.Close()
		if encoding == "br" {
			t.Fatalf("Accept-Encoding %q got brotli", accept)
		}
	}
}

func TestNegotiationServesRawWithoutAnAcceptEncoding(t *testing.T)

{
	mux, sizes := buildDir(t, allArtifacts)
	resp := request(t, mux, "/app.wasm", "")
	defer func() { _ = resp.Body.Close() }()
	if got := resp.Header.Get("Content-Encoding"); got != "" {
		t.Fatalf("Content-Encoding = %q, want none", got)
	}
	if got := resp.Header.Get("Content-Length"); got != strconv.Itoa(sizes["app.wasm"]) {
		t.Fatalf("Content-Length = %q, want the raw size", got)
	}
}

func TestNegotiationFallsThroughWhenNoArtifactExists(t *testing.T)

{
	mux, _ := buildDir(t, map[string]string{"app.wasm": "raw only"})
	resp := request(t, mux, "/app.wasm", "gzip, br")
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d", resp.StatusCode)
	}
	if got := resp.Header.Get("Content-Encoding"); got != "" {
		t.Fatalf("Content-Encoding = %q with no artifact on disk", got)
	}
}

func TestDirectArtifactURLsAreLabelled(t *testing.T)

{
	mux, _ := buildDir(t, allArtifacts)
	for path, want := range map[string]string{
		"/app.wasm.br": "br",
		"/app.wasm.gz": "gzip",
	} {
		resp := request(t, mux, path+"?v=abc123", "gzip, br")
		encoding := resp.Header.Get("Content-Encoding")
		contentType := resp.Header.Get("Content-Type")
		cache := resp.Header.Get("Cache-Control")
		vary := resp.Header.Get("Vary")
		_ = resp.Body.Close()
		if encoding != want {
			t.Errorf("%s Content-Encoding = %q, want %q", path, encoding, want)
		}
		if contentType != "application/wasm" {
			t.Errorf("%s Content-Type = %q", path, contentType)
		}
		if cache != "public, max-age=31536000, immutable" {
			t.Errorf("%s Cache-Control = %q", path, cache)
		}
		if vary != "Accept-Encoding" {
			t.Errorf("%s Vary = %q", path, vary)
		}
	}
}

func TestStampedPageRevalidates(t *testing.T)

{
	mux, _ := buildDir(t, map[string]string{"app.wasm": "raw"})
	for _, target := range []string{"/", "/index.html"} {
		resp := request(t, mux, target, "gzip")
		cache := resp.Header.Get("Cache-Control")
		_ = resp.Body.Close()
		if cache != "no-cache" {
			t.Errorf("%s Cache-Control = %q, want no-cache", target, cache)
		}
	}
}

func TestAcceptedEncodings(t *testing.T)

{
	cases := map[string]map[string]float64{
		"":                     nil,
		"gzip":                 {"gzip": 1},
		"gzip, deflate, br":    {"gzip": 1, "deflate": 1, "br": 1},
		"BR":                   {"br": 1},
		" gzip ;q=0.5 , br ":   {"gzip": 0.5, "br": 1},
		"br;q=0, gzip":         {"br": 0, "gzip": 1},
		"*":                    {"*": 1},
		"*, br;q=0":            {"*": 1, "br": 0},
		"identity;q=1, br;q=0": {"identity": 1, "br": 0},
		"br;q=bogus, gzip":     {"br": 0, "gzip": 1},
		"br;q=2, gzip":         {"br": 0, "gzip": 1},
	}
	for header, want := range cases {
		got := acceptedEncodings(header)
		if len(got) != len(want) {
			t.Errorf("acceptedEncodings(%q) = %v, want %v", header, got, want)
			continue
		}
		for name, quality := range want {
			if got[name] != quality {
				t.Errorf("acceptedEncodings(%q)[%q] = %v, want %v", header, name, got[name], quality)
			}
		}
	}
}

func TestNegotiationHonorsClientEncodingPreference(t *testing.T)

{
	mux, _ := buildDir(t, map[string]string{
		"app.wasm":    "raw",
		"app.wasm.br": "brotli",
		"app.wasm.gz": "gzip",
	})
	resp := request(t, mux, "/app.wasm", "gzip;q=1, br;q=0.1")
	defer func() { _ = resp.Body.Close() }()
	if got := resp.Header.Get("Content-Encoding"); got != "gzip" {
		t.Fatalf("Content-Encoding = %q, want gzip", got)
	}
}

func TestNegotiationUsesServerPreferenceForEqualQualities(t *testing.T)

{
	mux, _ := buildDir(t, map[string]string{
		"app.wasm":    "raw",
		"app.wasm.br": "brotli",
		"app.wasm.gz": "gzip",
	})
	resp := request(t, mux, "/app.wasm", "gzip, br")
	defer func() { _ = resp.Body.Close() }()
	if got := resp.Header.Get("Content-Encoding"); got != "br" {
		t.Fatalf("Content-Encoding = %q, want br", got)
	}
}

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,
		WriteTimeout:      10 * time.Second,
		OutboundQueueSize: 64,
		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.WriteTimeout > 0 {
			runtime.limits.WriteTimeout = limits.WriteTimeout
		}
		if limits.OutboundQueueSize > 0 {
			runtime.limits.OutboundQueueSize = limits.OutboundQueueSize
		}
		if limits.ResumeTTL > 0 {
			runtime.limits.ResumeTTL = limits.ResumeTTL
		}
		if limits.ReplayMessages > 0 {
			runtime.limits.ReplayMessages = limits.ReplayMessages
		}
	}
}

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 newStreamBusEndpoint(runtime *WSRuntime) *streamBusEndpoint

{
	maximum := DefaultSSCLimits().MaxMessageBytes
	if runtime != nil && runtime.limits.MaxMessageBytes > 0 {
		maximum = runtime.limits.MaxMessageBytes
	}
	return &streamBusEndpoint{
		runtime: runtime,
		bus: streambus.NewInMemory(streambus.Config{
			DefaultBuffer: 256, MaxBuffer: 4096, ReplayCapacity: 256,
			MaxPayloadBytes: maximum,
		}),
	}
}

func registerStreamBus(mux *http.ServeMux, runtime *WSRuntime)

{
	if !streamBusEnabled() {
		return
	}
	endpoint := newStreamBusEndpoint(runtime)
	mux.Handle(streamBusPath, runtime.Guard(endpoint))
	mux.HandleFunc("/__rfw/streambus-config", func(w http.ResponseWriter, _ *http.Request) {
		hash, _ := streamCertHash.Load().(string)
		port, _ := streamHTTP3Port.Load().(string)
		if hash == "" || port == "" {
			http.Error(w, "streambus certificate is not ready", http.StatusServiceUnavailable)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]string{"certificateHash": hash, "port": port})
	})
	streamEndpoints.Store(mux, endpoint)
}

func newStreamBusConnection(ctx context.Context, request *http.Request, transport *wt.Session, stream *wt.Stream, bus *streambus.InMemory, maximum int) (*streamBusConnection, error)

{
	topic := fmt.Sprintf("rfw/connection/%d", streamID.Add(1))
	subscription, err := bus.Subscribe(ctx, streambus.SubscribeOptions{Topic: topic, Buffer: 256, Overflow: streambus.Block})
	if err != nil {
		return nil, err
	}
	connection := &streamBusConnection{
		request: request, transport: transport, stream: stream, reader: bufio.NewReader(stream),
		bus: bus, topic: topic, subscription: subscription, maximum: maximum, done: make(chan struct{}),
	}
	go connection.writeLoop()
	return connection, nil
}

func streamBusProtocolLoop(connection *streamBusConnection, runtime *WSRuntime) error

{
	var session *Session
	var subscribed []string
	defer func() {
		streamClientsMu.Lock()
		for _, name := range subscribed {
			if set := streamClients[name]; set != nil {
				delete(set, connection)
				if len(set) == 0 {
					delete(streamClients, name)
				}
			}
		}
		streamClientsMu.Unlock()
		SuspendSession(session, runtime.ResumeTTL())
	}()

	for {
		var message Inbound
		if err := connection.Receive(&message); err != nil {
			return err
		}
		if session == nil {
			var resumed bool
			var err error
			session, resumed, err = runtime.OpenSession(connection.request, message.ResumeToken)
			if err != nil {
				_ = connection.Send(Outbound{Error: NewActionError("session_rejected", "session rejected")})
				return err
			}
			if !bindStreamBusConnection(connection, session, true) {
				_ = connection.Send(Outbound{Error: NewActionError("session_rejected", "session already has an active connection")})
				return errors.New("streambus: session already has an active connection")
			}
			if resumed {
				replayStreamBus(connection, session, message.Ack)
			} else if message.ResumeToken != "" {
				sendStreamBusSession(connection, session, Outbound{Control: "resume_rejected", Error: NewActionError("resume_rejected", "session could not be resumed")})
			}
		}
		session.Acknowledge(message.Ack)
		if message.Control == "ping" {
			sendStreamBusSession(connection, session, Outbound{Control: "pong"})
			continue
		}
		if err := session.AcceptInbound(message.Sequence); err != nil {
			if errors.Is(err, ErrDuplicateMessage) {
				continue
			}
			sendStreamBusSession(connection, session, Outbound{ID: message.ID, Action: message.Action, Error: NewActionError("sequence_gap", "client message sequence gap")})
			continue
		}
		if !session.AllowMessage(runtime.MessagesPerMinute()) {
			sendStreamBusSession(connection, session, Outbound{ID: message.ID, Action: message.Action, Error: NewActionError("rate_limited", "message rate limit exceeded")})
			continue
		}
		authorizeCtx, cancel := runtime.HandlerContext(context.Background())
		authorizeErr := runtime.Authorize(authorizeCtx, session, message)
		cancel()
		if authorizeErr != nil {
			sendStreamBusSession(connection, session, Outbound{Component: message.Component, Action: message.Action, ID: message.ID, Error: NewActionError("forbidden", "message forbidden")})
			continue
		}
		if message.Action != "" {
			payload, actionErr := runtime.DispatchAction(context.Background(), session, message)
			sendStreamBusSession(connection, session, Outbound{Action: message.Action, ID: message.ID, Payload: payload, Error: actionErr})
			continue
		}
		if message.Component != "" && message.Payload != nil && message.Payload["unsubscribe"] == true {
			streamClientsMu.Lock()
			if set := streamClients[message.Component]; set != nil {
				delete(set, connection)
				if len(set) == 0 {
					delete(streamClients, message.Component)
				}
			}
			for index, name := range subscribed {
				if name == message.Component {
					subscribed = append(subscribed[:index], subscribed[index+1:]...)
					break
				}
			}
			streamClientsMu.Unlock()
			sendStreamBusSession(connection, session, Outbound{Component: message.Component, Control: "unsubscribed"})
			continue
		}
		if component, ok := Get(message.Component); ok {
			streamClientsMu.Lock()
			if streamClients[message.Component] == nil {
				streamClients[message.Component] = make(map[*streamBusConnection]*Session)
			}
			if _, tracked := streamClients[message.Component][connection]; !tracked {
				streamClients[message.Component][connection] = session
				subscribed = append(subscribed, message.Component)
			}
			streamClientsMu.Unlock()
			response := component.HandleWithSession(session, message.Payload)
			if response != nil {
				switch value := response.(type) {
				case *InitSnapshot:
					if value != nil {
						sendStreamBusSession(connection, session, Outbound{Component: message.Component, ID: message.ID, Payload: map[string]any{"initSnapshot": value}})
					}
				case InitSnapshot:
					sendStreamBusSession(connection, session, Outbound{Component: message.Component, ID: message.ID, Payload: map[string]any{"initSnapshot": value}})
				default:
					sendStreamBusSession(connection, session, Outbound{Component: message.Component, ID: message.ID, Payload: response})
				}
				continue
			}
			if message.Payload != nil && message.Payload["init"] == true {
				sendStreamBusSession(connection, session, Outbound{Component: message.Component, Payload: map[string]any{"session": session.ID()}})
				continue
			}
		}
		sendStreamBusSession(connection, session, Outbound{Control: "ack"})
	}
}

func bindStreamBusConnection(connection *streamBusConnection, session *Session, handoff bool) bool

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

func sendStreamBusSession(connection *streamBusConnection, session *Session, out Outbound)

{
	if session == nil {
		return
	}
	session.outboundMu.Lock()
	defer session.outboundMu.Unlock()
	if session.streamConnection != connection {
		return
	}
	_ = connection.Send(session.PrepareOutbound(out))
}

func replayStreamBus(connection *streamBusConnection, session *Session, acknowledged uint64)

{
	session.outboundMu.Lock()
	defer session.outboundMu.Unlock()
	if session.streamConnection != connection {
		return
	}
	messages, err := session.ReplayAfter(acknowledged)
	if err != nil {
		_ = connection.Send(session.PrepareOutbound(Outbound{Error: NewActionError("resync_required", "message history is no longer available")}))
		return
	}
	for _, message := range messages {
		_ = connection.Send(message)
	}
}

func startStreamBusHTTP3(addr string, mux *http.ServeMux, tlsConfig *tls.Config)

{
	value, ok := streamEndpoints.Load(mux)
	if !ok {
		return
	}
	endpoint := value.(*streamBusEndpoint)
	port := strings.TrimPrefix(addr, ":")
	if _, parsedPort, err := net.SplitHostPort(addr); err == nil {
		port = parsedPort
	}
	streamHTTP3Port.Store(port)
	if len(tlsConfig.Certificates) > 0 && len(tlsConfig.Certificates[0].Certificate) > 0 {
		hash := sha256.Sum256(tlsConfig.Certificates[0].Certificate[0])
		streamCertHash.Store(fmt.Sprintf("%x", hash[:]))
	}
	h3 := &http3.Server{
		Addr: addr, Handler: mux, TLSConfig: tlsConfig.Clone(), EnableDatagrams: true,
		QUICConfig: &quic.Config{EnableDatagrams: true, EnableStreamResetPartialDelivery: true},
	}
	server := &wt.Server{H3: h3}
	wt.ConfigureHTTP3Server(h3)
	endpoint.mu.Lock()
	endpoint.server = server
	endpoint.mu.Unlock()
	go func() {
		if err := server.ListenAndServe(); err != nil {
			log.Printf("streambus HTTP/3 server: %v", err)
		}
	}()
}

func configureConnectionWriter(ws *websocket.Conn, writeTimeout time.Duration, queueSize int)

{
	if ws == nil || queueSize <= 0 {
		return
	}
	writer := &connectionWriter{
		ws:           ws,
		writeTimeout: writeTimeout,
		queue:        make(chan []Outbound, queueSize),
		resync:       make(chan struct{}, 1),
		done:         make(chan struct{}),
	}
	if _, loaded := connWriters.LoadOrStore(ws, writer); loaded {
		return
	}
	go writer.run()
}

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 AnswerControl(ws, msg) {
			continue
		}
		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})
	}

	streamClientsMu.RLock()
	streamTargets := make([]struct {
		connection *streamBusConnection
		session    *Session
	}, 0, len(streamClients[name]))
	for connection, session := range streamClients[name] {
		streamTargets = append(streamTargets, struct {
			connection *streamBusConnection
			session    *Session
		}{connection: connection, session: session})
	}
	streamClientsMu.RUnlock()
	for _, target := range streamTargets {
		if options.Session != "" && target.session.ID() != options.Session {
			continue
		}
		sendStreamBusSession(target.connection, target.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
	}
	if writerValue, ok := connWriters.Load(ws); ok {
		writerValue.(*connectionWriter).enqueue(func() ([]Outbound, error) {
			messages, err := session.ReplayAfter(acknowledged)
			if err != nil {
				return []Outbound{session.PrepareOutbound(Outbound{
					Error: NewActionError("resync_required", "message history is no longer available"),
				})}, nil
			}
			return messages, nil
		})
		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"),
		}), DefaultSSCLimits().WriteTimeout)
		return
	}
	for _, message := range messages {
		if err := sendOutboundUnlocked(ws, message, DefaultSSCLimits().WriteTimeout); err != nil {
			log.Printf("send replay: %v", err)
			_ = ws.Close()
			return
		}
	}
}

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
	}
	out, err := prepareOutboundPayload(out)
	if err != nil {
		log.Printf("marshal outbound payload: %v", err)
		return
	}
	if writerValue, ok := connWriters.Load(ws); ok {
		writerValue.(*connectionWriter).enqueue(func() ([]Outbound, error) {
			return []Outbound{session.PrepareOutbound(out)}, nil
		})
		return
	}
	lock := connectionWriteLock(ws)
	lock.Lock()
	defer lock.Unlock()
	if err := sendOutboundUnlocked(ws, session.PrepareOutbound(out), DefaultSSCLimits().WriteTimeout); err != nil {
		log.Printf("send: %v", err)
		_ = ws.Close()
	}
}

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.streamConnection != nil || (session.connectionManaged && !handoff) {
		return false, false
	}
	session.connection = ws
	session.streamConnection = nil
	return true, false
}

func SendOutbound(ws *websocket.Conn, out Outbound)

{
	if writerValue, ok := connWriters.Load(ws); ok {
		writerValue.(*connectionWriter).enqueue(func() ([]Outbound, error) {
			prepared, err := prepareOutboundPayload(out)
			if err != nil {
				return nil, err
			}
			return []Outbound{prepared}, nil
		})
		return
	}
	lock := connectionWriteLock(ws)
	lock.Lock()
	defer lock.Unlock()
	var err error
	out, err = prepareOutboundPayload(out)
	if err != nil {
		log.Printf("marshal outbound payload: %v", err)
		return
	}
	if err := sendOutboundUnlocked(ws, out, DefaultSSCLimits().WriteTimeout); err != nil {
		log.Printf("send: %v", err)
		_ = ws.Close()
	}
}

func sendOutboundUnlocked(ws *websocket.Conn, out Outbound, writeTimeout time.Duration) error

{
	b, err := marshalOutbound(out)
	if err != nil {
		return err
	}
	if writeTimeout > 0 {
		if err := ws.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {
			return err
		}
	}
	if err := websocket.Message.Send(ws, b); err != nil {
		return err
	}
	if writeTimeout > 0 {
		return ws.SetWriteDeadline(time.Time{})
	}
	return nil
}

func prepareOutboundPayload(out Outbound) (Outbound, error)

{
	if out.Payload == nil || out.encodedPayload != nil {
		return out, nil
	}
	payload, err := json.Marshal(out.Payload)
	if err != nil {
		return out, err
	}
	out.encodedPayload = payload
	return out, nil
}

func openBlockedProtocolSocket(t *testing.T, runtime *WSRuntime) (*websocket.Conn, func())

{
	t.Helper()
	serverConn, clientConn := net.Pipe()
	listener := &singleConnListener{conn: serverConn, closed: make(chan struct{})}
	server := &http.Server{
		Handler: runtime.Guard(websocket.Handler(func(ws *websocket.Conn) {
			wsHandler(ws, runtime)
		})),
		ReadHeaderTimeout: time.Second,
	}
	serveDone := make(chan error, 1)
	go func() { serveDone <- server.Serve(listener) }()

	config, err := websocket.NewConfig("ws://pipe.invalid/ws", "http://pipe.invalid")
	if err != nil {
		t.Fatalf("configure pipe websocket: %v", err)
	}
	client, err := websocket.NewClient(config, clientConn)
	if err != nil {
		_ = server.Close()
		_ = listener.Close()
		t.Fatalf("open pipe websocket: %v", err)
	}
	cleanup := func() {
		_ = client.SetDeadline(time.Now())
		_ = client.Close()
		_ = server.Close()
		_ = listener.Close()
		select {
		case err := <-serveDone:
			if err != nil && !errors.Is(err, http.ErrServerClosed) && !errors.Is(err, net.ErrClosed) {
				t.Errorf("serve pipe websocket: %v", err)
			}
		case <-time.After(time.Second):
			t.Error("pipe websocket server did not stop")
		}
	}
	return client, cleanup
}

func waitForConnectionCount(t *testing.T, component string, count int) map[*websocket.Conn]*Session

{
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		connMu.RLock()
		current := make(map[*websocket.Conn]*Session, len(connections[component]))
		for ws, session := range connections[component] {
			current[ws] = session
		}
		connMu.RUnlock()
		if len(current) == count {
			return current
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatalf("component %q connection count did not reach %d", component, count)
	return nil
}

func waitForSessionDetached(t *testing.T, session *Session)

{
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		session.deliveryMu.Lock()
		attached := session.attached
		session.deliveryMu.Unlock()
		if !attached {
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatal("session did not detach after its connection closed")
}

func TestWriteDeadlineClosesAClientThatDoesNotRead(t *testing.T)

{
	const componentName = "write-deadline-client"
	Register(NewHostComponent(componentName, func(map[string]any) any {
		return map[string]any{"ready": true}
	}))
	runtime := NewWSRuntime(WithSSCLimits(SSCLimits{
		WriteTimeout:      25 * time.Millisecond,
		OutboundQueueSize: 4,
	}))
	client, cleanup := openBlockedProtocolSocket(t, runtime)
	defer cleanup()

	sendTestInbound(t, client, Inbound{Component: componentName, Payload: map[string]any{"init": true}})
	targets := waitForConnectionCount(t, componentName, 1)
	var session *Session
	for _, candidate := range targets {
		session = candidate
	}
	waitForSessionDetached(t, session)

	if err := client.SetReadDeadline(time.Now().Add(time.Second)); err == nil {
		var raw []byte
		if err := websocket.Message.Receive(client, &raw); err == nil {
			t.Fatalf("slow client remained open after the write deadline: %s", raw)
		}
	} else if !errors.Is(err, io.ErrClosedPipe) && !errors.Is(err, net.ErrClosed) {
		t.Fatalf("set client read deadline: %v", err)
	}
	ReleaseSession(session)
}

func TestInvalidPayloadDoesNotCloseConnectionOrConsumeSequence(t *testing.T)

{
	client, server, closeSockets := openWriteTestSocket(t)
	defer closeSockets()
	configureConnectionWriter(server, time.Second, 4)
	session := newSession("invalid-payload")
	BindSessionConnection(server, session)

	SendSessionOutbound(server, session, Outbound{Payload: make(chan int)})
	SendSessionOutbound(server, session, Outbound{Payload: "valid"})
	message := receiveOrderedMessage(t, client)
	if message.Sequence != 1 || message.Payload != "valid" {
		t.Fatalf("valid message after marshal failure = %#v", message)
	}
}

func TestBroadcastDoesNotWaitForSlowClientAndOverflowResumes(t *testing.T)

{
	const componentName = "bounded-broadcast-client"
	Register(NewHostComponent(componentName, func(map[string]any) any {
		return map[string]any{"ready": true}
	}))
	slowRuntime := NewWSRuntime(WithSSCLimits(SSCLimits{
		WriteTimeout:      2 * time.Second,
		OutboundQueueSize: 2,
		ResumeTTL:         time.Second,
		ReplayMessages:    16,
	}))
	fastRuntime := NewWSRuntime(WithSSCLimits(SSCLimits{
		WriteTimeout:      2 * time.Second,
		OutboundQueueSize: 16,
		ResumeTTL:         time.Second,
		ReplayMessages:    16,
	}))

	slowClient, closeSlow := openBlockedProtocolSocket(t, slowRuntime)
	defer closeSlow()
	sendTestInbound(t, slowClient, Inbound{Component: componentName, Payload: map[string]any{"init": true}})
	slowTargets := waitForConnectionCount(t, componentName, 1)
	var (
		slowServer  *websocket.Conn
		slowSession *Session
	)
	for ws, session := range slowTargets {
		slowServer = ws
		slowSession = session
	}
	token := slowSession.ResumeToken()
	sessionID := slowSession.ID()
	if token == "" {
		t.Fatal("slow session was not resumable")
	}

	deadline := time.Now().Add(time.Second)
	for {
		slowSession.deliveryMu.Lock()
		sequence := slowSession.outboundSeq
		slowSession.deliveryMu.Unlock()
		writerValue, ok := connWriters.Load(slowServer)
		if ok && sequence == 1 && len(writerValue.(*connectionWriter).queue) == 0 {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("slow client writer did not enter its blocked write")
		}
		time.Sleep(time.Millisecond)
	}

	server, wsURL := openRuntimeServer(t, fastRuntime)
	defer server.Close()
	fastClient, err := websocket.Dial(wsURL, "", server.URL)
	if err != nil {
		t.Fatalf("dial fast websocket: %v", err)
	}
	defer closeTestResource(t, fastClient)
	sendTestInbound(t, fastClient, Inbound{Component: componentName, Payload: map[string]any{"init": true}})
	_ = receiveOrderedMessage(t, fastClient)
	waitForConnectionCount(t, componentName, 2)

	started := time.Now()
	for tick := 1; tick <= 3; tick++ {
		Broadcast(componentName, map[string]any{"tick": tick})
	}
	if elapsed := time.Since(started); elapsed > 100*time.Millisecond {
		t.Fatalf("broadcast waited for the slow client: %s", elapsed)
	}
	for tick := 1; tick <= 3; tick++ {
		message := receiveOrderedMessage(t, fastClient)
		payload, ok := message.Payload.(map[string]any)
		if !ok || payload["tick"] != float64(tick) {
			t.Fatalf("fast client broadcast %d = %#v", tick, message)
		}
	}

	if err := slowClient.SetReadDeadline(time.Now().Add(time.Second)); err != nil {
		t.Fatalf("set slow client read deadline: %v", err)
	}
	resyncRequired := false
	for {
		var raw []byte
		if err := websocket.Message.Receive(slowClient, &raw); err != nil {
			break
		}
		var message Outbound
		if err := json.Unmarshal(raw, &message); err != nil {
			t.Fatalf("decode slow client message: %v", err)
		}
		if message.Control == "resync_required" && message.Error != nil && message.Error.Code == "resync_required" {
			resyncRequired = true
		}
	}
	if !resyncRequired {
		t.Fatal("queue overflow closed the slow client without resync_required")
	}
	waitForSessionDetached(t, slowSession)

	resumedClient, err := websocket.Dial(wsURL, "", server.URL)
	if err != nil {
		t.Fatalf("dial resumed websocket: %v", err)
	}
	sendTestInbound(t, resumedClient, Inbound{Control: "ping", ResumeToken: token})
	for sequence := uint64(1); sequence <= 4; sequence++ {
		message := receiveOrderedMessage(t, resumedClient)
		if message.Session != sessionID || message.Sequence != sequence {
			t.Fatalf("resumed message %d = %#v", sequence, message)
		}
		if sequence >= 2 {
			payload, ok := message.Payload.(map[string]any)
			if !ok || payload["tick"] != float64(sequence-1) {
				t.Fatalf("replayed broadcast %d = %#v", sequence-1, message)
			}
		}
	}
	pong := receiveOrderedMessage(t, resumedClient)
	if pong.Control != "pong" || pong.Sequence != 0 {
		t.Fatalf("resume did not continue on the new connection: %#v", pong)
	}
	closeTestResource(t, resumedClient)
	waitForSessionDetached(t, slowSession)
	ReleaseSession(slowSession)
}

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 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
}