hostclient

func TestHandleHostPayloadMismatchTriggersResync(t *testing.T)

{
	root := newFakeRoot()
	root.elems["greeting"] = &fakeElement{
		exists:   true,
		expected: encodeExpectation("server"),
		text:     "tampered",
	}

	payload := map[string]any{"greeting": "fresh"}
	mismatches := handleHostPayload(root, payload, nil)
	if len(mismatches) != 1 {
		t.Fatalf("expected 1 mismatch, got %d", len(mismatches))
	}
	if root.elems["greeting"].text != "tampered" {
		t.Fatalf("text was updated despite mismatch")
	}
	resync := buildResyncPayload(mismatches)
	body, ok := resync["resync"].(map[string]any)
	if !ok {
		t.Fatalf("resync payload missing body")
	}
	if body["reason"] != "host-var-mismatch" {
		t.Fatalf("unexpected reason %v", body["reason"])
	}
	vars, ok := body["vars"].([]map[string]string)
	if ok {
		if vars[0]["var"] != "greeting" {
			t.Fatalf("unexpected var name %s", vars[0]["var"])
		}
		if vars[0]["expected"] == vars[0]["actualHash"] {
			t.Fatalf("expected hashes to differ on mismatch")
		}
	}
}

func TestLegacyExpectationRequiresResync(t *testing.T)

{
	root := newFakeRoot()
	root.elems["greeting"] = &fakeElement{
		exists:   true,
		expected: "sha1:2b42fba6b3f0c7b0d352c30b63f055c1b2f507a2",
		text:     "hello",
	}

	if mismatches := handleHostPayload(root, map[string]any{"greeting": "updated"}, nil); len(mismatches) != 1 {
		t.Fatalf("legacy expectation was trusted without verification: %+v", mismatches)
	}
}

func TestInitSnapshotRecoveryAndUpdate(t *testing.T)

{
	root := newFakeRoot()
	root.elems["count"] = &fakeElement{
		exists:   true,
		expected: encodeExpectation("1"),
		text:     "0",
	}

	if mismatches := handleHostPayload(root, map[string]any{"count": "2"}, nil); len(mismatches) == 0 {
		t.Fatalf("expected mismatch when expectation diverges")
	}

	snapHTML := `<span data-host-var="count" data-host-expected="` + encodeExpectation("1") + `">1</span>`
	applyInitSnapshot(root, &initSnapshotPayload{HTML: snapHTML})

	if mismatches := handleHostPayload(root, map[string]any{"count": "3"}, nil); len(mismatches) != 0 {
		t.Fatalf("expected clean hydration after snapshot")
	}

	elem := root.HostVar("count").(*fakeElement)
	if elem.text != "3" {
		t.Fatalf("expected text to update to 3, got %s", elem.text)
	}
	if elem.expected != encodeExpectation("3") {
		t.Fatalf("expected hash to reflect new value")
	}
}

func updateHostVar(root componentRoot, name, value string) *hydrationMismatch

{
	node := root.HostVar(name)
	if !node.Exists() {
		return nil
	}
	expectedAttr := node.Attr(hostExpectedAttr)
	actualText := node.Text()
	matches, alg, actualHash := expectationMatches(expectedAttr, actualText)
	if !matches {
		return &hydrationMismatch{
			VarName:     name,
			Expected:    expectedAttr,
			Actual:      actualText,
			ActualHash:  actualHash,
			ExpectedAlg: alg,
		}
	}
	node.SetText(value)
	node.SetAttr(hostExpectedAttr, encodeExpectation(value))
	return nil
}

func handleHostPayload(root componentRoot, payload map[string]any, updateSignal func(name string, raw any)) []hydrationMismatch

{
	mismatches := make([]hydrationMismatch, 0)
	for key, raw := range payload {
		if key == "initSnapshot" || strings.HasPrefix(key, "_") {
			continue
		}
		mismatch := updateHostVar(root, key, fmt.Sprintf("%v", raw))
		if mismatch != nil {
			mismatches = append(mismatches, *mismatch)
		}
		if updateSignal != nil {
			updateSignal(key, raw)
		}
	}
	return mismatches
}

func buildResyncPayload(mismatches []hydrationMismatch) map[string]any

{
	entries := make([]map[string]string, 0, len(mismatches))
	for _, m := range mismatches {
		entries = append(entries, map[string]string{
			"var":         m.VarName,
			"expected":    m.Expected,
			"expectedAlg": m.ExpectedAlg,
			"actual":      m.Actual,
			"actualHash":  m.ActualHash,
		})
	}
	return map[string]any{
		"resync": map[string]any{
			"reason": "host-var-mismatch",
			"vars":   entries,
		},
	}
}

func decodeInitSnapshotPayload(raw any) *initSnapshotPayload

{
	if raw == nil {
		return nil
	}
	m, ok := raw.(map[string]any)
	if !ok {
		return nil
	}
	html, _ := m["html"].(string)
	if html == "" {
		return nil
	}
	var vars []string
	if list, ok := m["vars"].([]any); ok {
		vars = make([]string, 0, len(list))
		for _, item := range list {
			if s, ok := item.(string); ok {
				vars = append(vars, s)
			}
		}
	} else if list, ok := m["vars"].([]string); ok {
		vars = append(vars, list...)
	}
	return &initSnapshotPayload{HTML: html, Vars: vars}
}

func init()

{
	cb = fnres.NewCircuitBreaker(5, 30*time.Second)
	cb.OnStateChange(func(from, to fnres.State) {
		if debug {
			log.Printf("hostclient: circuit %v -> %v", from, to)
		}
	})
	hydrateCB = fnres.NewCircuitBreaker(3, 15*time.Second)
	sendCache = fncaching.NewInMemory[string](
		fncaching.WithMaxEntries[string](256),
		fncaching.WithTTL[string](5*time.Second),
	)
}

func hostWSURL() string

{
	if u := js.Get("RFW_HOST_URL"); u.Truthy() {
		if s := normalizeWSURL(u.String()); s != "" {
			return s
		}
	}
	host := js.Location().Get("host").String()
	if h := js.Get("RFW_HOST"); h.Truthy() {
		host = h.String()
	}
	return normalizeWSURL(host)
}

func normalizeWSURL(raw string) string

{
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return ""
	}
	switch {
	case strings.HasPrefix(raw, "ws://"), strings.HasPrefix(raw, "wss://"):
	case strings.HasPrefix(raw, "http://"):
		raw = "ws://" + strings.TrimPrefix(raw, "http://")
	case strings.HasPrefix(raw, "https://"):
		raw = "wss://" + strings.TrimPrefix(raw, "https://")
	default:
		scheme := "wss"
		if js.Location().Get("protocol").String() == "http:" {
			scheme = "ws"
		}
		raw = scheme + "://" + raw
	}
	if rest := raw[strings.Index(raw, "://")+3:]; !strings.Contains(rest, "/") {
		raw = strings.TrimRight(raw, "/") + "/ws"
	}
	return raw
}

func connectionLoop()

{
	for {
		url := hostWSURL()
		connectionState.Set(ConnectionConnecting)

		err := fnres.Retry(context.Background(), func() error {
			return cb.Execute(func() error {
				if debug {
					log.Printf("hostclient: dialing %s", url)
				}
				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
				defer cancel()
				c, _, derr := websocket.Dial(ctx, url, nil)
				if derr != nil {
					return derr
				}
				c.SetReadLimit(maxInboundMessageBytes)

				sendMu.Lock()
				mu.Lock()
				conn = c
				pend := pending
				pending = nil
				mu.Unlock()

				if debug {
					log.Printf("hostclient: connected")
				}
				connectionState.Set(ConnectionConnected)

				mu.RLock()
				names := make([]string, 0, len(bindings)+len(handlers))
				for name := range bindings {
					names = append(names, name)
				}
				for name := range handlers {
					if _, bound := bindings[name]; !bound {
						names = append(names, name)
					}
				}
				mu.RUnlock()
				deliveryMu.Lock()
				unacknowledged := make([]message, 0, len(outbox))
				for sequence := uint64(1); sequence <= nextOutbound; sequence++ {
					if msg, ok := outbox[sequence]; ok {
						unacknowledged = append(unacknowledged, msg)
					}
				}
				deliveryMu.Unlock()
				initialized := make(map[string]struct{})
				for _, msg := range unacknowledged {
					sendMessageUnlocked(c, msg)
					if name, ok := initMessageName(msg); ok {
						initialized[name] = struct{}{}
					}
				}
				for _, msg := range pend {
					sendMessageUnlocked(c, msg)
					if name, ok := initMessageName(msg); ok {
						initialized[name] = struct{}{}
					}
				}
				for _, name := range names {
					if _, sent := initialized[name]; sent {
						continue
					}
					sendMessageUnlocked(c, message{name: name, payload: map[string]any{"init": true}})
				}
				sendMu.Unlock()

				ctx2, cancel2 := context.WithCancel(context.Background())
				defer cancel2()
				errCh := make(chan error, 2)
				go func() { errCh <- guardedLoop("host read loop", func() error { return readLoop(ctx2, c) }) }()
				go func() { errCh <- guardedLoop("host ping loop", func() error { return pingLoop(ctx2, c) }) }()
				loopErr := <-errCh
				cancel2()
				closeErr := c.Close(websocket.StatusInternalError, "connection closed")

				mu.Lock()
				conn = nil
				mu.Unlock()
				connectionState.Set(ConnectionDisconnected)
				if loopErr != nil {
					return loopErr
				}
				return closeErr
			})
		},
			fnres.WithAttempts(5),
			fnres.WithDelay(time.Second, 30*time.Second),
			fnres.WithFactor(2),
			fnres.WithJitter(0.1),
			fnres.WithRetryIf(func(err error) bool { return err != nil }),
		)

		if err != nil && debug {
			log.Printf("hostclient: connection attempt failed: %v", err)
		}
		connectionState.Set(ConnectionDisconnected)

		// Back off before reconnecting to avoid tight loops on persistent failures.
		time.Sleep(time.Second)
	}
}

func pingLoop(ctx context.Context, c *websocket.Conn) error

{
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			pctx, cancel := context.WithTimeout(ctx, 5*time.Second)
			err := c.Ping(pctx)
			cancel()
			if err != nil {
				return err
			}
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

func readLoop(ctx context.Context, c *websocket.Conn) error

{
	for {
		var msg struct {
			Component   string       `json:"component"`
			Action      string       `json:"action"`
			Control     string       `json:"control"`
			ID          string       `json:"id"`
			Payload     any          `json:"payload"`
			Error       *ActionError `json:"error"`
			Session     string       `json:"session"`
			Sequence    uint64       `json:"sequence"`
			Ack         uint64       `json:"ack"`
			ResumeToken string       `json:"resumeToken"`
		}
		if err := wsjson.Read(ctx, c, &msg); err != nil {
			return err
		}
		if debug {
			log.Printf("hostclient: recv %s %v", msg.Component, msg.Payload)
		}
		prepareInboundDelivery(msg.Session, msg.Control)
		deliveryMu.Lock()
		for sequence := range outbox {
			if sequence <= msg.Ack {
				delete(outbox, sequence)
			}
		}
		if msg.Sequence != 0 {
			if msg.Sequence <= lastInbound {
				deliveryMu.Unlock()
				continue
			}
			if lastInbound != 0 && msg.Sequence != lastInbound+1 {
				deliveryMu.Unlock()
				connectionState.Set(ConnectionDesynced)
				return errors.New("hostclient: server message sequence gap")
			}
			lastInbound = msg.Sequence
		}
		if msg.ResumeToken != "" {
			resumeToken = msg.ResumeToken
		}
		deliveryMu.Unlock()
		if msg.ID != "" {
			callMu.Lock()
			replyChannel := pendingCalls[msg.ID]
			if replyChannel != nil {
				delete(pendingCalls, msg.ID)
			}
			callMu.Unlock()
			if replyChannel != nil {
				replyChannel <- actionReply{payload: msg.Payload, err: msg.Error}
				continue
			}
		}
		if msg.Control != "" {
			continue
		}
		payload, _ := msg.Payload.(map[string]any)
		if payload == nil {
			payload = make(map[string]any)
		}
		mu.RLock()
		h, hasHandler := handlers[msg.Component]
		b, hasBinding := bindings[msg.Component]
		mu.RUnlock()
		if hasHandler {
			if msg.Session != "" {
				payload["_session"] = msg.Session
			}
			js.Guard("host handler: "+msg.Component, func() { h(payload) })
			continue
		}
		if hasBinding {
			js.Guard("host binding: "+msg.Component, func() {
				applyHostBinding(msg.Component, payload, b)
			})
		}
	}
}

func applyHostBinding(component string, payload map[string]any, binding componentBinding)

{
	rootEl := dom.ComponentRoot(binding.id)
	if !rootEl.Truthy() {
		return
	}
	root := newComponentRoot(rootEl)
	if snap := decodeInitSnapshotPayload(payload["initSnapshot"]); snap != nil {
		applyInitSnapshot(root, snap)
		if len(snap.Vars) > 0 {
			binding.vars = append([]string(nil), snap.Vars...)
			mu.Lock()
			bindings[component] = binding
			mu.Unlock()
		}
		return
	}

	mismatches := handleHostPayload(root, payload, func(name string, raw any) {
		signals := dom.SnapshotComponentSignals(binding.id)
		if signals == nil {
			return
		}
		if signal, ok := signals[name]; ok {
			if setter, ok := signal.(interface{ SetFromHost(any) }); ok {
				setter.SetFromHost(raw)
			}
		}
	})
	if len(mismatches) == 0 {
		return
	}
	for _, mismatch := range mismatches {
		log.Printf("hostclient: hydration mismatch component=%s var=%s expected=%s actualHash=%s actual=%q", component, mismatch.VarName, mismatch.Expected, mismatch.ActualHash, mismatch.Actual)
	}
	resyncErr := hydrateCB.Execute(func() error {
		Send(component, buildResyncPayload(mismatches))
		return nil
	})
	if resyncErr != nil {
		log.Printf("hostclient: hydration circuit open, skipping resync for %s", component)
	}
}

func prepareInboundDelivery(remoteSession, control string)

{
	sessionMu.Lock()
	previousSession := sessionID
	if remoteSession != "" {
		sessionID = remoteSession
	}
	sessionMu.Unlock()
	if control != "resume_rejected" && (remoteSession == "" || previousSession == "" || remoteSession == previousSession) {
		return
	}
	deliveryMu.Lock()
	lastInbound = 0
	resumeToken = ""
	deliveryMu.Unlock()
}

func RegisterComponent(id, name string, vars []string)

{
	mu.Lock()
	bindings[name] = componentBinding{id: id, vars: vars}
	current := conn
	if current == nil {
		pending = append(pending, message{name: name, payload: map[string]any{"init": true}})
	}
	mu.Unlock()
	connect()
	if current != nil {
		sendMessage(current, message{name: name, payload: map[string]any{"init": true}})
	}
}

func Send(name string, payload any)

{
	connect()
	if dedupEnabled(name) {
		key := fmt.Sprintf("%s|%v", name, payload)
		if _, ok, _ := sendCache.Get(context.Background(), key); ok {
			return
		}
		if err := sendCache.Set(context.Background(), key, "sent", 5*time.Second); err != nil {
			log.Printf("hostclient: dedup cache set failed: %v", err)
		}
	}

	mu.RLock()
	c := conn
	mu.RUnlock()
	if c == nil {
		mu.Lock()
		pending = append(pending, message{name: name, payload: payload})
		mu.Unlock()
		return
	}
	if debug {
		log.Printf("hostclient: send %s %v", name, payload)
	}
	sendMessage(c, message{name: name, payload: payload})
}

func RegisterHandler(name string, h func(map[string]any)) func()

{
	token := handlerSequence.Add(1)
	mu.Lock()
	handlers[name] = h
	handlerTokens[name] = token
	current := conn
	if current == nil {
		pending = append(pending, message{name: name, payload: map[string]any{"init": true}})
	}
	mu.Unlock()
	connect()
	if current != nil {
		sendMessage(current, message{name: name, payload: map[string]any{"init": true}})
	}
	var once sync.Once
	return func() {
		once.Do(func() {
			mu.Lock()
			if handlerTokens[name] != token {
				mu.Unlock()
				return
			}
			delete(handlers, name)
			delete(handlerTokens, name)
			filtered := pending[:0]
			for _, queued := range pending {
				if queued.name == name && isInitPayload(queued.payload) {
					continue
				}
				filtered = append(filtered, queued)
			}
			pending = filtered
			current := conn
			mu.Unlock()

			unsubscribe := message{name: name, payload: map[string]any{"unsubscribe": true}}
			if current != nil {
				sendMessage(current, unsubscribe)
				return
			}
			mu.Lock()
			pending = append(pending, unsubscribe)
			mu.Unlock()
		})
	}
}

func sendMessageUnlockedWithWriter(c *websocket.Conn, msg message, writer messageWriter)

{
	deliveryMu.Lock()
	if msg.sequence == 0 {
		nextOutbound++
		msg.sequence = nextOutbound
		outbox[msg.sequence] = msg
	}
	token := resumeToken
	ack := lastInbound
	deliveryMu.Unlock()
	outbound := wireMessage{
		Component:   msg.name,
		Action:      msg.action,
		ID:          msg.id,
		Payload:     msg.payload,
		Sequence:    msg.sequence,
		Ack:         ack,
		ResumeToken: token,
	}
	ctx := context.Background()
	_ = writer(ctx, c, outbound)
}

func Call[Request, Response any](ctx context.Context, action string, request Request) (Response, error)

{
	var zero Response
	if ctx == nil {
		ctx = context.Background()
	}
	if action == "" {
		return zero, errors.New("hostclient: empty action name")
	}
	connect()
	id := fmt.Sprintf("call-%d", callSequence.Add(1))
	replyChannel := make(chan actionReply, 1)
	callMu.Lock()
	pendingCalls[id] = replyChannel
	callMu.Unlock()

	msg := message{action: action, id: id, payload: request}
	mu.RLock()
	current := conn
	mu.RUnlock()
	if current == nil {
		mu.Lock()
		pending = append(pending, msg)
		mu.Unlock()
	} else {
		sendMessage(current, msg)
	}

	select {
	case reply := <-replyChannel:
		if reply.err != nil {
			return zero, reply.err
		}
		data, err := json.Marshal(reply.payload)
		if err != nil {
			return zero, fmt.Errorf("hostclient: encode action response: %w", err)
		}
		if err := json.Unmarshal(data, &zero); err != nil {
			return zero, fmt.Errorf("hostclient: decode action response: %w", err)
		}
		return zero, nil
	case <-ctx.Done():
		callMu.Lock()
		delete(pendingCalls, id)
		callMu.Unlock()
		return zero, ctx.Err()
	}
}

func TestGuardedLoopConvertsPanicAndNextLoopRuns(t *testing.T)

{
	previous := js.OnRuntimePanic
	defer func() { js.OnRuntimePanic = previous }()
	recovered := 0
	js.OnRuntimePanic = func(any, string, []byte) { recovered++ }

	if err := guardedLoop("read", func() error { panic("bad push") }); err == nil {
		t.Fatal("panicking loop returned nil")
	}
	if err := guardedLoop("read", func() error { return nil }); err != nil {
		t.Fatalf("next loop returned %v", err)
	}
	if recovered != 1 {
		t.Fatalf("recovered loop panics = %d, want 1", recovered)
	}
}

func TestRegisterHandlerUnsubscribeQueuesWireUnsubscribe(t *testing.T)

{
	name := "scoped-handler"
	before := pendingCount()
	unsubscribe := RegisterHandler(name, func(map[string]any) {})
	if got := pendingCount() - before; got != 1 {
		t.Fatalf("queued subscribe messages = %d, want 1", got)
	}
	unsubscribe()

	mu.RLock()
	_, stillRegistered := handlers[name]
	queued := append([]message(nil), pending...)
	mu.RUnlock()
	if stillRegistered {
		t.Fatal("handler remained registered after unsubscribe")
	}
	count := 0
	for _, item := range queued {
		if item.name == name {
			values, _ := item.payload.(map[string]any)
			if values["unsubscribe"] == true {
				count++
			}
			if values["init"] == true {
				t.Fatal("stale init remained queued after unsubscribe")
			}
		}
	}
	if count != 1 {
		t.Fatalf("queued unsubscribe messages = %d, want 1", count)
	}
}

func TestStaleUnsubscribeDoesNotRemoveReplacementHandler(t *testing.T)

{
	name := "replacement-handler"
	first := RegisterHandler(name, func(map[string]any) {})
	second := RegisterHandler(name, func(map[string]any) {})
	first()
	mu.RLock()
	_, registered := handlers[name]
	mu.RUnlock()
	if !registered {
		t.Fatal("stale unsubscribe removed replacement handler")
	}
	second()
}

func TestSendMessageSerializesSequenceAndWrite(t *testing.T)

{
	deliveryMu.Lock()
	savedNext := nextOutbound
	savedOutbox := outbox
	nextOutbound = 0
	outbox = map[uint64]message{}
	deliveryMu.Unlock()
	defer func() {
		deliveryMu.Lock()
		nextOutbound = savedNext
		outbox = savedOutbox
		deliveryMu.Unlock()
	}()

	firstEntered := make(chan struct{}, 1)
	firstRelease := make(chan struct{})
	secondEntered := make(chan struct{}, 1)
	firstDone := make(chan struct{})
	secondDone := make(chan struct{})
	firstWriter := func(_ context.Context, _ *websocket.Conn, message wireMessage) error {
		firstEntered <- struct{}{}
		<-firstRelease
		if message.Sequence != 1 {
			t.Errorf("first sequence = %d, want 1", message.Sequence)
		}
		return nil
	}
	secondWriter := func(_ context.Context, _ *websocket.Conn, message wireMessage) error {
		secondEntered <- struct{}{}
		if message.Sequence != 2 {
			t.Errorf("second sequence = %d, want 2", message.Sequence)
		}
		return nil
	}

	go func() {
		sendMessageWithWriter(nil, message{name: "first"}, firstWriter)
		close(firstDone)
	}()
	<-firstEntered
	go func() {
		sendMessageWithWriter(nil, message{name: "second"}, secondWriter)
		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
}

func TestPrepareInboundDeliveryResetsNewSessionState(t *testing.T)

{
	sessionMu.Lock()
	savedSession := sessionID
	sessionID = "old-session"
	sessionMu.Unlock()
	deliveryMu.Lock()
	savedInbound := lastInbound
	savedToken := resumeToken
	lastInbound = 9
	resumeToken = "old-token"
	deliveryMu.Unlock()
	defer func() {
		sessionMu.Lock()
		sessionID = savedSession
		sessionMu.Unlock()
		deliveryMu.Lock()
		lastInbound = savedInbound
		resumeToken = savedToken
		deliveryMu.Unlock()
	}()

	prepareInboundDelivery("new-session", "")

	sessionMu.RLock()
	currentSession := sessionID
	sessionMu.RUnlock()
	deliveryMu.Lock()
	currentInbound := lastInbound
	currentToken := resumeToken
	deliveryMu.Unlock()
	if currentSession != "new-session" || currentInbound != 0 || currentToken != "" {
		t.Fatalf("delivery state was not reset: session=%q inbound=%d token=%q", currentSession, currentInbound, currentToken)
	}
}

func TestPrepareInboundDeliveryResetsRejectedResume(t *testing.T)

{
	sessionMu.Lock()
	savedSession := sessionID
	sessionID = "current-session"
	sessionMu.Unlock()
	deliveryMu.Lock()
	savedInbound := lastInbound
	savedToken := resumeToken
	lastInbound = 9
	resumeToken = "old-token"
	deliveryMu.Unlock()
	defer func() {
		sessionMu.Lock()
		sessionID = savedSession
		sessionMu.Unlock()
		deliveryMu.Lock()
		lastInbound = savedInbound
		resumeToken = savedToken
		deliveryMu.Unlock()
	}()

	prepareInboundDelivery("current-session", "resume_rejected")

	deliveryMu.Lock()
	currentInbound := lastInbound
	currentToken := resumeToken
	deliveryMu.Unlock()
	if currentInbound != 0 || currentToken != "" {
		t.Fatalf("rejected resume state was not reset: inbound=%d token=%q", currentInbound, currentToken)
	}
}