Serve T5DE-hosted rooms; a second client instance's own data

Hosted rooms (internal/hosted): room calls for them answered here, their IMQ queues served (a new
proxy.IMQRouter hook takes them out of subscriptions to IMVU), chat relayed through the hub's
WebSocket (hub.Client gains Send, OnRoom and OnConnect), the room list extended, room products
checked against the real owner, and anything else naming a hosted room kept from IMVU.

T5DE_INSTANCE gives a second client's proxy its own data folder and log.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

README.md

@@ -80,6 +80,16 @@ that aren't here yet (up to 15 seconds) so they can be shown and tried on. Buyin
 (`api/service/shop/svc_catalog_purchase.php`) adds them to the person's hub library; IMVU products in the same
 purchase go on to IMVU without them.
 
+T5DE-hosted rooms (`internal/hosted`, t5de-hub's `docs/hosted-rooms.md`): rooms living on the hub, with instance IDs
+`2100000000-<n>` and chat IDs `2100000000+n`. The proxy answers the client's room calls for them (`getRoomCardInfo`,
+`getOrMakeChat`, `getRoomState`, `setRoomState`, `getParticipants`, `terminateChat`), serves their IMQ queues
+(`/chat/<id>`, `inv:/room/room-<id>`, `room_occupancy/<id>`: subscriptions answered here through `proxy.IMQRouter`,
+chat to and from the hub over its WebSocket), adds them to the room list, checks the room's products against its real
+owner (`authorizeProducts2`), and answers any other request to IMVU naming one ("not found") rather than send it.
+
+`T5DE_INSTANCE=<name>` (set by the client's `t5de-instances` patch for a second client run beside the first) keeps
+the proxy's data in `%APPDATA%\T5DE-<name>\proxy` and its log in `t5de-proxy-<name>.log`.
+
 For finding out what IMVU's APIs send, `T5DE_PROXY_DUMP=<text>` logs the request and response bodies of requests
 whose path contains it (test accounts only: bodies hold personal data). In Git Bash, keep the value from starting
 with `/` (or set `MSYS_NO_PATHCONV=1`), or it's turned into a Windows path.

cmd/t5de-proxy/main.go

@@ -17,6 +17,7 @@ import (
 	"crypto/rand"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -29,12 +30,15 @@ import (
 	"path/filepath"
 	"runtime"
 	"strings"
+	"sync"
 	"time"
 
 	"git.t5de.dev/t5de/t5de-proxy/internal/buildinfo"
 	"git.t5de.dev/t5de/t5de-proxy/internal/control"
 	"git.t5de.dev/t5de/t5de-proxy/internal/fetch"
+	"git.t5de.dev/t5de/t5de-proxy/internal/hosted"
 	"git.t5de.dev/t5de/t5de-proxy/internal/hub"
+	"git.t5de.dev/t5de/t5de-proxy/internal/imq"
 	"git.t5de.dev/t5de/t5de-proxy/internal/parent"
 	"git.t5de.dev/t5de/t5de-proxy/internal/products"
 	"git.t5de.dev/t5de/t5de-proxy/internal/proxy"
@@ -55,7 +59,7 @@ func main() {
 	trace := flag.Bool("trace", false, "log full URLs and every IMQ frame")
 	packs := flag.String("packs", "", "community product packs: directories of packs, separated by "+string(filepath.ListSeparator)+
 		" (default: %APPDATA%/T5DE/packs and $T5DE_PACKS)")
-	data := flag.String("data", "", "where the proxy keeps its state (default: %APPDATA%/T5DE/proxy)")
+	data := flag.String("data", "", "where the proxy keeps its state (default: %APPDATA%/T5DE/proxy, or T5DE-<n> for $T5DE_INSTANCE)")
 	version := flag.Bool("version", false, "print the version and exit")
 	flag.Parse()
 	if *version {
@@ -65,6 +69,10 @@ func main() {
 
 	var out io.Writer = os.Stderr
 	if *logFile != "" {
+		if instance := os.Getenv("T5DE_INSTANCE"); instance != "" {
+			// Another client beside the usual one (its T5DE_INSTANCE): its own log.
+			*logFile = strings.TrimSuffix(*logFile, ".log") + "-" + instance + ".log"
+		}
 		f, err := os.Create(*logFile)
 		if err != nil {
 			os.Exit(1)
@@ -96,6 +104,9 @@ func dataDir(flagValue string) string {
 		return flagValue
 	}
 	if config, err := os.UserConfigDir(); err == nil {
+		if instance := os.Getenv("T5DE_INSTANCE"); instance != "" {
+			return filepath.Join(config, "T5DE-"+instance, "proxy") // the client's own data (its T5DE_INSTANCE)
+		}
 		return filepath.Join(config, "T5DE", "proxy")
 	}
 	return "."
@@ -138,10 +149,25 @@ func run(logger *log.Logger, listen, ready string, parentPID int, trace bool, pa
 		handler.Interceptor, handler.IMQFilter = prods, prods
 		logger.Printf("community products from %s", strings.Join(packs, ", "))
 	}
+	// T5DE-hosted rooms (internal/hosted), served with the hub the proxy is signed in to.
+	hubNow := &hubRef{}
+	rooms := &hosted.Rooms{Hub: hubNow, Logf: logger.Printf}
+	if prods != nil {
+		prods.Hosted = rooms
+		rooms.User, rooms.LocalProduct, rooms.GlobalProduct = prods.User, prods.LocalID, prods.ProductID
+		prods.HostedOwner = rooms.Owner
+		rooms.Deliver = func(c *proxy.IMQConn, m imq.Delivered) {
+			m.Message = prods.Incoming(c, m)
+			c.Deliver(m)
+		}
+	}
 	// The hub connection, which the client's T5DE settings can sign in and out (internal/control).
 	manager := &control.Manager{Path: filepath.Join(data, "hub.json"), DefaultHub: envOr("T5DE_HUB", defaultHub),
 		Log: logger, OpenBrowser: openBrowser}
 	manager.Attach = func(client *hub.Client) {
+		hubNow.set(client)
+		client.OnRoom, client.OnConnect = rooms.FromHub, rooms.Reconnected
+		rooms.OnJoin, rooms.OnLeave = client.Join, client.Leave
 		fetcher := fetch.New(client, hubProducts, logger.Printf)
 		client.OnLibrary = func() { fetcher.SyncLibrary(context.Background()) }
 		if prods == nil {
@@ -168,7 +194,7 @@ func run(logger *log.Logger, listen, ready string, parentPID int, trace bool, pa
 		}
 	}
 	manager.Start()
-	chain := proxy.Interceptors{manager}
+	chain := proxy.Interceptors{manager, rooms} // hosted rooms' calls before the products'
 	if prods != nil {
 		chain = append(chain, prods)
 	}
@@ -186,6 +212,38 @@ func run(logger *log.Logger, listen, ready string, parentPID int, trace bool, pa
 	return server.Serve(ln)
 }
 
+// hubRef is the hub client the proxy is signed in with, for the hosted rooms (none while signed out).
+type hubRef struct {
+	mu sync.Mutex
+	c  *hub.Client
+}
+
+func (h *hubRef) set(c *hub.Client) {
+	h.mu.Lock()
+	h.c = c
+	h.mu.Unlock()
+}
+
+func (h *hubRef) get() *hub.Client {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return h.c
+}
+
+func (h *hubRef) Send(v any) bool {
+	if c := h.get(); c != nil {
+		return c.Send(v)
+	}
+	return false
+}
+
+func (h *hubRef) Call(ctx context.Context, method, path string, body, out any) error {
+	if c := h.get(); c != nil {
+		return c.Call(ctx, method, path, body, out)
+	}
+	return errors.New("not signed in to T5DE's hub")
+}
+
 func writeReady(path, address, token string) error {
 	data, err := json.Marshal(map[string]string{"address": address, "token": token})
 	if err != nil {

internal/hosted/hosted.go

@@ -0,0 +1,827 @@
+// Package hosted runs T5DE-hosted rooms for the client (t5de-hub's docs/hosted-rooms.md): rooms that live on T5DE's
+// hub, which the client joins and uses like IMVU's rooms. The proxy answers the client's room calls for them
+// (XML-RPC), serves their IMQ queues (subscriptions answered here, messages to and from the hub), and adds them to
+// the room list. IMVU never hears of them.
+//
+// A hosted room has a number n; its instance ID is "2100000000-<n>" and its chat ID 2100000000+n.
+package hosted
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"regexp"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"git.t5de.dev/t5de/t5de-proxy/internal/imq"
+	"git.t5de.dev/t5de/t5de-proxy/internal/proxy"
+	"git.t5de.dev/t5de/t5de-proxy/internal/xmlrpc"
+)
+
+// Base is the first part of hosted rooms' instance IDs, and the start of their chat IDs.
+const Base = 2100000000
+
+// Hub is what the rooms need of the hub: its realtime connection and its API.
+type Hub interface {
+	Send(v any) bool
+	Call(ctx context.Context, method, path string, body, out any) error
+}
+
+// Room is a hosted room as the hub lists it.
+type Room struct {
+	N           int64  `json:"n"`
+	InstanceID  string `json:"instanceId"`
+	ChatID      int64  `json:"chatId"`
+	Name        string `json:"name"`
+	Description string `json:"description"`
+	RoomPid     int64  `json:"roomPid"`
+	Rating      string `json:"rating"`
+	MaxUsers    int    `json:"maxUsers"`
+	OwnerImvuID int64  `json:"ownerImvuId"`
+	OwnerName   string `json:"ownerName"`
+	Occupancy   int    `json:"occupancy"`
+}
+
+// Participant is someone in a hosted room.
+type Participant struct {
+	ImvuID int64 `json:"imvu_id"`
+	Seat   int   `json:"seat"`
+}
+
+// Rooms serves hosted rooms to the client.
+type Rooms struct {
+	Hub  Hub
+	Logf func(format string, args ...any)
+	// User is the IMVU account the client is signed in as (0 if not yet known).
+	User func() int64
+	// Deliver gives the client a message from someone in a room (through the products' Incoming).
+	Deliver func(c *proxy.IMQConn, m imq.Delivered)
+	// OnJoin and OnLeave report being in a room's chat (the room channel, for others' community products).
+	OnJoin  func(chat, room string)
+	OnLeave func(chat string)
+	// LocalProduct and GlobalProduct translate scenes' community products between the hub's IDs and the client's
+	// local IDs (false: not a community product, or not here).
+	LocalProduct  func(hubID string) (int64, bool)
+	GlobalProduct func(local int64) (string, bool)
+
+	mu      sync.Mutex
+	rooms   map[int64]*room
+	waiting map[int64][]chan joinAnswer
+	owners  map[int64]int64 // room -> its owner's IMVU account, as last looked up
+	last    int64           // the room the client joined last
+}
+
+type room struct {
+	chat         int64
+	instance     string
+	participants []Participant
+	sequence     uint64
+	conns        map[*proxy.IMQConn]bool // subscribed to the room's chat queue
+}
+
+type joinAnswer struct {
+	Type         string        `json:"type"`
+	Room         int64         `json:"room"`
+	Chat         int64         `json:"chat"`
+	Participants []Participant `json:"participants"`
+	Sequence     uint64        `json:"sequence"`
+	Reason       string        `json:"reason"`
+	Message      string        `json:"message"`
+}
+
+var instancePattern = regexp.MustCompile(`^` + strconv.Itoa(Base) + `-([1-9][0-9]{0,8})$`)
+
+// RoomNumber is the hosted room an instance ID names.
+func RoomNumber(instance string) (int64, bool) {
+	m := instancePattern.FindStringSubmatch(instance)
+	if m == nil {
+		return 0, false
+	}
+	n, err := strconv.ParseInt(m[1], 10, 64)
+	return n, err == nil && n < 2147483647-Base
+}
+
+// ChatRoom is the hosted room a chat ID belongs to.
+func ChatRoom(chat int64) (int64, bool) {
+	if chat <= Base || chat > 2147483647 {
+		return 0, false
+	}
+	return chat - Base, true
+}
+
+func instanceOf(n int64) string { return fmt.Sprintf("%d-%d", Base, n) }
+
+// queueRoom is the hosted room a queue belongs to, and which of its queues it is: "chat", "node" (the room's
+// inv:/room/room-<instance>) or "occupancy" (room_occupancy/<instance>).
+func queueRoom(queue string) (int64, string, bool) {
+	switch {
+	case strings.HasPrefix(queue, "/chat/"):
+		chat, err := strconv.ParseInt(strings.TrimPrefix(queue, "/chat/"), 10, 64)
+		if err != nil {
+			return 0, "", false
+		}
+		n, ok := ChatRoom(chat)
+		return n, "chat", ok
+	case strings.HasPrefix(queue, "inv:/room/room-"):
+		n, ok := RoomNumber(strings.TrimPrefix(queue, "inv:/room/room-"))
+		return n, "node", ok
+	case strings.HasPrefix(queue, "room_occupancy/"):
+		n, ok := RoomNumber(strings.TrimPrefix(queue, "room_occupancy/"))
+		return n, "occupancy", ok
+	}
+	return 0, "", false
+}
+
+// IsHostedQueue reports whether a queue is a hosted room's.
+func IsHostedQueue(queue string) bool {
+	_, _, ok := queueRoom(queue)
+	return ok
+}
+
+func (r *Rooms) logf(format string, args ...any) {
+	if r.Logf != nil {
+		r.Logf(format, args...)
+	}
+}
+
+func (r *Rooms) user() int64 {
+	if r.User == nil {
+		return 0
+	}
+	return r.User()
+}
+
+// --- the hub's messages ---
+
+// FromHub handles a "room.*" message from the hub.
+func (r *Rooms) FromHub(data []byte) {
+	var m struct {
+		joinAnswer
+		From    int64  `json:"from"`
+		Mount   string `json:"mount"`
+		Message string `json:"message"`
+	}
+	if json.Unmarshal(data, &m) != nil {
+		return
+	}
+	switch m.Type {
+	case "room.joined", "room.error":
+		m.joinAnswer.Message = m.Message // the outer field takes "message"
+		r.mu.Lock()
+		if m.Type == "room.joined" {
+			rm := r.roomLocked(m.Room)
+			rm.chat, rm.participants, rm.sequence = m.Chat, m.Participants, m.Sequence
+		}
+		waiting := r.waiting[m.Room]
+		delete(r.waiting, m.Room)
+		r.mu.Unlock()
+		for _, w := range waiting {
+			w <- m.joinAnswer
+		}
+		if m.Type == "room.error" {
+			r.logf("hosted: room %d: %s (%s)", m.Room, m.Reason, m.Message)
+		}
+	case "room.participants":
+		r.mu.Lock()
+		if rm, ok := r.rooms[m.Room]; ok {
+			rm.participants = m.Participants
+		}
+		r.mu.Unlock()
+	case "room.message":
+		r.mu.Lock()
+		rm, ok := r.rooms[m.Room]
+		var conns []*proxy.IMQConn
+		chat := Base + m.Room
+		if ok {
+			if m.Sequence > rm.sequence {
+				rm.sequence = m.Sequence
+			}
+			for c := range rm.conns {
+				conns = append(conns, c)
+			}
+		}
+		r.mu.Unlock()
+		for _, c := range conns {
+			d := imq.Delivered{UserID: fmt.Sprintf("/user/%d", m.From), Queue: fmt.Sprintf("/chat/%d", chat),
+				Mount: m.Mount, Message: []byte(m.Message), Sequence: m.Sequence}
+			if r.Deliver != nil {
+				r.Deliver(c, d)
+			} else {
+				c.Deliver(d)
+			}
+		}
+	case "room.closed":
+		r.mu.Lock()
+		delete(r.rooms, m.Room)
+		r.mu.Unlock()
+		if r.OnLeave != nil {
+			r.OnLeave(strconv.FormatInt(Base+m.Room, 10))
+		}
+		r.logf("hosted: room %d was closed", m.Room)
+	}
+}
+
+func (r *Rooms) roomLocked(n int64) *room {
+	if r.rooms == nil {
+		r.rooms = map[int64]*room{}
+	}
+	rm, ok := r.rooms[n]
+	if !ok {
+		rm = &room{chat: Base + n, instance: instanceOf(n), conns: map[*proxy.IMQConn]bool{}}
+		r.rooms[n] = rm
+	}
+	return rm
+}
+
+// Reconnected joins the rooms the client is in again (the hub forgets them when the connection drops).
+func (r *Rooms) Reconnected() {
+	r.mu.Lock()
+	var ns []int64
+	for n := range r.rooms {
+		ns = append(ns, n)
+	}
+	r.mu.Unlock()
+	for _, n := range ns {
+		r.Hub.Send(map[string]any{"type": "room.join", "room": n})
+	}
+}
+
+// join asks the hub to let the client into a room, and waits for the answer.
+func (r *Rooms) join(ctx context.Context, n int64) (joinAnswer, error) {
+	w := make(chan joinAnswer, 1)
+	r.mu.Lock()
+	if r.waiting == nil {
+		r.waiting = map[int64][]chan joinAnswer{}
+	}
+	r.waiting[n] = append(r.waiting[n], w)
+	r.mu.Unlock()
+	if !r.Hub.Send(map[string]any{"type": "room.join", "room": n}) {
+		return joinAnswer{}, fmt.Errorf("the hub isn't connected")
+	}
+	select {
+	case a := <-w:
+		return a, nil
+	case <-ctx.Done():
+		return joinAnswer{}, fmt.Errorf("the hub didn't answer")
+	}
+}
+
+func (r *Rooms) leave(n int64) {
+	r.mu.Lock()
+	_, in := r.rooms[n]
+	delete(r.rooms, n)
+	r.mu.Unlock()
+	if in {
+		r.Hub.Send(map[string]any{"type": "room.leave", "room": n})
+		if r.OnLeave != nil {
+			r.OnLeave(strconv.FormatInt(Base+n, 10))
+		}
+	}
+}
+
+// --- IMQ ---
+
+// Route takes hosted rooms' queues out of a subscription and answers the client for them (proxy.IMQRouter).
+func (r *Rooms) Route(c *proxy.IMQConn, unsubscribe bool, s imq.Subscription) imq.Subscription {
+	var rest imq.Subscription
+	for _, q := range s.Queues {
+		if !r.routeQueue(c, unsubscribe, q, 0, false) {
+			rest.Queues = append(rest.Queues, q)
+		}
+	}
+	for _, q := range s.WithResults {
+		if !r.routeQueue(c, unsubscribe, q.Queue, q.OpID, true) {
+			rest.WithResults = append(rest.WithResults, q)
+		}
+	}
+	return rest
+}
+
+func (r *Rooms) routeQueue(c *proxy.IMQConn, unsubscribe bool, queue string, opID uint64, withResult bool) bool {
+	n, kind, ok := queueRoom(queue)
+	if !ok {
+		return false
+	}
+	me := fmt.Sprintf("/user/%d", r.user())
+	if unsubscribe {
+		if kind == "chat" {
+			r.mu.Lock()
+			if rm, ok := r.rooms[n]; ok {
+				delete(rm.conns, c)
+			}
+			r.mu.Unlock()
+		}
+		if withResult {
+			c.Send(imq.G2cResult, imq.Result(opID, 0))
+		}
+		return true
+	}
+	switch kind {
+	case "chat":
+		r.mu.Lock()
+		rm := r.roomLocked(n)
+		rm.conns[c] = true
+		seq := rm.sequence
+		r.mu.Unlock()
+		c.Send(imq.G2cJoinedQueue, imq.JoinedQueue(me, queue, seq))
+		c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "control", imq.MessageMount, nil, seq))
+		c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "messages", imq.MessageMount, nil, seq))
+		c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "participants", imq.StateMount, nil, seq))
+	case "node":
+		c.Send(imq.G2cJoinedQueue, imq.JoinedQueue(me, queue, 0))
+		c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "node", imq.MessageMount, nil, 0))
+	case "occupancy":
+		c.Send(imq.G2cJoinedQueue, imq.JoinedQueue(me, queue, 0))
+		c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "occupants", imq.StateMount, nil, 0))
+	}
+	if withResult {
+		c.Send(imq.G2cResult, imq.Result(opID, 0))
+	}
+	return true
+}
+
+// Closed forgets a connection's subscriptions.
+func (r *Rooms) Closed(c *proxy.IMQConn) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for _, rm := range r.rooms {
+		delete(rm.conns, c)
+	}
+}
+
+// Send sends what the client said or did in a hosted room to the hub (the client's message has already had its
+// community products taken out, as in IMVU's rooms). It reports whether the queue was a hosted room's.
+func (r *Rooms) Send(queue, mount string, message []byte) bool {
+	n, kind, ok := queueRoom(queue)
+	if !ok {
+		return false
+	}
+	if kind != "chat" {
+		return true // nothing to say on the others
+	}
+	if mount == "messages" {
+		// Sitting down: "*seat <n>" in the chat text.
+		var m struct {
+			Message string `json:"message"`
+		}
+		if json.Unmarshal(message, &m) == nil && strings.HasPrefix(m.Message, "*seat ") {
+			if seat, err := strconv.Atoi(strings.TrimSpace(strings.TrimPrefix(m.Message, "*seat "))); err == nil && seat >= 0 && seat < 100 {
+				r.Hub.Send(map[string]any{"type": "room.seat", "room": n, "seat": seat})
+			}
+		}
+	}
+	r.Hub.Send(map[string]any{"type": "room.send", "room": n, "mount": mount, "message": string(message)})
+	return true
+}
+
+// --- HTTP: XML-RPC, the participant post and the room list ---
+
+var participantPath = regexp.MustCompile(`^/chat/chat-([0-9]+)/participants/`)
+
+// Intercept answers the client's calls about hosted rooms (proxy.Interceptor). It goes before the products'.
+func (r *Rooms) Intercept(x *proxy.Exchange) error {
+	host, path := strings.ToLower(x.URL.Hostname()), x.URL.Path
+	if !strings.HasSuffix(host, ".imvu.com") {
+		return nil
+	}
+	if host == "api.imvu.com" {
+		if m := participantPath.FindStringSubmatch(path); m != nil {
+			if chat, err := strconv.ParseInt(m[1], 10, 64); err == nil {
+				if _, ok := ChatRoom(chat); ok {
+					x.Response = proxy.LocalResponse(200, "application/json", []byte(`{"status":"success"}`))
+				}
+			}
+		}
+		if x.Response == nil {
+			r.keepFromIMVU(x)
+		}
+		return nil
+	}
+	switch path {
+	case "/api/xmlrpc/chat.php", "/api/xmlrpc/client.php":
+		if err := r.interceptXMLRPC(x); err != nil || x.Response != nil {
+			return err
+		}
+	case "/api/rooms/rooms_list_paginated.php":
+		r.addToRoomList(x)
+		return nil
+	case "/api/rooms/room_rating.php":
+		if _, ok := RoomNumber(x.URL.Query().Get("room_instance_id")); ok {
+			x.Response = proxy.LocalResponse(200, "application/json", []byte(`{"score":0,"result":"success"}`))
+			return nil
+		}
+	}
+	r.keepFromIMVU(x)
+	return nil
+}
+
+// keepFromIMVU answers, rather than sends, any other request to IMVU naming a hosted room (its instance ID, or the
+// chat ID of a room the client is in): IMVU doesn't know them, and needn't hear of them.
+func (r *Rooms) keepFromIMVU(x *proxy.Exchange) {
+	text := x.URL.Path + "?" + x.URL.RawQuery + " " + string(x.Body)
+	named := strings.Contains(text, strconv.Itoa(Base)+"-")
+	if !named {
+		r.mu.Lock()
+		for _, rm := range r.rooms {
+			if strings.Contains(text, strconv.FormatInt(rm.chat, 10)) {
+				named = true
+				break
+			}
+		}
+		r.mu.Unlock()
+	}
+	if !named {
+		return
+	}
+	r.logf("hosted: answered a request to %s%s naming a hosted room (not found)", x.URL.Host, x.URL.Path)
+	x.Response = proxy.LocalResponse(404, "application/json", []byte(`{"status":"failure","error":"NOT_FOUND"}`))
+}
+
+func xmlrpcAnswer(v xmlrpc.Value) *http.Response {
+	return proxy.LocalResponse(200, "text/xml", (&xmlrpc.Response{Params: []xmlrpc.Value{v}}).Marshal())
+}
+
+func xmlrpcFault(message string) *http.Response {
+	return proxy.LocalResponse(200, "text/xml", (&xmlrpc.Response{Fault: &xmlrpc.Fault{Code: 1, String: message}}).Marshal())
+}
+
+func member(s xmlrpc.Struct, name string) string {
+	v, ok := s.Get(name)
+	if !ok {
+		return ""
+	}
+	switch v := v.(type) {
+	case string:
+		return v
+	case int64:
+		return strconv.FormatInt(v, 10)
+	case float64:
+		return strconv.FormatFloat(v, 'f', -1, 64)
+	}
+	return ""
+}
+
+func (r *Rooms) interceptXMLRPC(x *proxy.Exchange) error {
+	call, err := xmlrpc.ParseCall(x.Body)
+	if err != nil || len(call.Params) == 0 {
+		return nil
+	}
+	args, _ := call.Params[0].(xmlrpc.Struct)
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+	switch call.Method {
+	case "test.getRoomCardInfo", "test.getRoomState", "test.setRoomState":
+		n, ok := RoomNumber(member(args, "room_instance_id"))
+		if !ok {
+			return nil
+		}
+		switch call.Method {
+		case "test.getRoomCardInfo":
+			x.Response = r.roomCard(ctx, n)
+		case "test.getRoomState":
+			x.Response = r.roomState(ctx, n)
+		default:
+			x.Response = r.saveRoomState(ctx, n, args)
+		}
+	case "chat.getOrMakeChat":
+		n, ok := RoomNumber(strings.TrimPrefix(member(args, "activity"), "publicroom-"))
+		if !ok {
+			return nil
+		}
+		x.Response = r.joinChat(ctx, n)
+	case "chat.getParticipants", "chat.terminateChat":
+		chat, err := strconv.ParseInt(member(args, "chatId"), 10, 64)
+		if err != nil {
+			return nil
+		}
+		n, ok := ChatRoom(chat)
+		if !ok {
+			return nil
+		}
+		if call.Method == "chat.terminateChat" {
+			r.leave(n)
+			x.Response = xmlrpcAnswer(true)
+			return nil
+		}
+		x.Response = xmlrpcAnswer(r.others(n))
+	}
+	return nil
+}
+
+func (r *Rooms) getRoom(ctx context.Context, n int64) (*Room, [][3]any, string, error) {
+	var out struct {
+		Room  Room `json:"room"`
+		Scene struct {
+			Contents [][3]any `json:"contents"`
+			Hash     string   `json:"hash"`
+		} `json:"scene"`
+	}
+	if err := r.Hub.Call(ctx, "GET", fmt.Sprintf("/api/v1/hosted-rooms/%d", n), nil, &out); err != nil {
+		return nil, nil, "", err
+	}
+	r.mu.Lock()
+	if r.owners == nil {
+		r.owners = map[int64]int64{}
+	}
+	r.owners[n] = out.Room.OwnerImvuID
+	r.mu.Unlock()
+	return &out.Room, out.Scene.Contents, out.Scene.Hash, nil
+}
+
+// Owner is the IMVU account owning the hosted room the client joined last (0 if none, or not known): the client
+// checks the room's products against it (products.HostedOwner).
+func (r *Rooms) Owner() int64 {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if _, in := r.rooms[r.last]; !in {
+		return 0
+	}
+	return r.owners[r.last]
+}
+
+func (r *Rooms) roomCard(ctx context.Context, n int64) *http.Response {
+	rm, _, _, err := r.getRoom(ctx, n)
+	if err != nil {
+		r.logf("hosted: room %d: %v", n, err)
+		return xmlrpcFault("This T5DE room isn't available.")
+	}
+	adult := rm.Rating == "adult"
+	return xmlrpcAnswer(xmlrpc.Struct{
+		{Name: "owner", Value: plain(rm.OwnerName)}, {Name: "owner_cid", Value: rm.OwnerImvuID},
+		{Name: "size", Value: int64(rm.Occupancy)}, {Name: "name", Value: plain(rm.Name)},
+		{Name: "lieutenants", Value: []xmlrpc.Value{}}, {Name: "roomOwners", Value: []xmlrpc.Value{}},
+		{Name: "allowApProducts", Value: adult}, {Name: "allowRoomShellChange", Value: false},
+		{Name: "allowRoomShellOverwrite", Value: false}, {Name: "allowJoinNewRoom", Value: false},
+		{Name: "allowLoadNewRoom", Value: false}, {Name: "maxOccupancy", Value: int64(rm.MaxUsers)},
+		{Name: "description", Value: plain(rm.Description)}, {Name: "imageUrl", Value: ""}, {Name: "language", Value: ""},
+		{Name: "isAP", Value: adult}, {Name: "isVIP", Value: false}, {Name: "allowGames", Value: false},
+		{Name: "isFavorite", Value: false}, {Name: "showFavorite", Value: false}, {Name: "showRatings", Value: false},
+		{Name: "enableFlagging", Value: false}, {Name: "private", Value: false},
+		{Name: "autoBootWhenOwnerLeaves", Value: false}, {Name: "rating", Value: float64(0)},
+		{Name: "permission", Value: ""}, {Name: "themedStatus", Value: ""}, {Name: "themeId", Value: ""},
+		{Name: "eligibleForThemedRoomSubmission", Value: ""}, {Name: "occupants", Value: []xmlrpc.Value{}},
+	})
+}
+
+func (r *Rooms) roomState(ctx context.Context, n int64) *http.Response {
+	rm, contents, hash, err := r.getRoom(ctx, n)
+	if err != nil {
+		r.logf("hosted: room %d: %v", n, err)
+		return xmlrpcFault("This T5DE room isn't available.")
+	}
+	var items []xmlrpc.Value
+	for _, it := range contents {
+		furni, ok1 := it[0].(float64)
+		placement, ok2 := it[2].(string)
+		if !ok1 || !ok2 {
+			continue
+		}
+		var product int64
+		switch p := it[1].(type) {
+		case float64:
+			product = int64(p)
+		case string:
+			local, ok := int64(0), false
+			if r.LocalProduct != nil {
+				local, ok = r.LocalProduct(p)
+			}
+			if !ok {
+				continue // a community product that isn't here
+			}
+			product = local
+		default:
+			continue
+		}
+		items = append(items, []xmlrpc.Value{int64(furni), product, placement})
+	}
+	if items == nil {
+		items = []xmlrpc.Value{}
+	}
+	return xmlrpcAnswer(xmlrpc.Struct{
+		{Name: "room_state", Value: xmlrpc.Struct{
+			{Name: "room_info", Value: xmlrpc.Struct{
+				{Name: "customers_id", Value: strconv.FormatInt(rm.OwnerImvuID, 10)},
+				{Name: "room_instance_id", Value: instanceOf(n)},
+				{Name: "room_pid", Value: strconv.FormatInt(rm.RoomPid, 10)},
+			}},
+			{Name: "room_contents", Value: items},
+		}},
+		{Name: "room_hash", Value: hash},
+	})
+}
+
+func (r *Rooms) saveRoomState(ctx context.Context, n int64, args xmlrpc.Struct) *http.Response {
+	v, _ := args.Get("room_contents")
+	list, _ := v.([]xmlrpc.Value)
+	contents := [][3]any{}
+	for _, item := range list {
+		it, ok := item.([]xmlrpc.Value)
+		if !ok || len(it) < 3 {
+			continue
+		}
+		furni, ok1 := asInt(it[0])
+		product, ok2 := asInt(it[1])
+		placement, ok3 := it[2].(string)
+		if !ok1 || !ok2 || !ok3 {
+			continue
+		}
+		var p any = product
+		if product >= 2000000000 {
+			id, ok := "", false
+			if r.GlobalProduct != nil {
+				id, ok = r.GlobalProduct(product)
+			}
+			if !ok {
+				continue
+			}
+			p = id
+		}
+		contents = append(contents, [3]any{furni, p, placement})
+	}
+	if err := r.Hub.Call(ctx, "PUT", fmt.Sprintf("/api/v1/hosted-rooms/%d/scene", n), map[string]any{"contents": contents}, nil); err != nil {
+		r.logf("hosted: saving room %d: %v", n, err)
+		return xmlrpcFault("Couldn't save the room on T5DE's hub.")
+	}
+	return xmlrpcAnswer(true)
+}
+
+func asInt(v xmlrpc.Value) (int64, bool) {
+	switch v := v.(type) {
+	case int64:
+		return v, true
+	case float64:
+		return int64(v), v == float64(int64(v))
+	case string:
+		n, err := strconv.ParseInt(v, 10, 64)
+		return n, err == nil
+	}
+	return 0, false
+}
+
+func (r *Rooms) joinChat(ctx context.Context, n int64) *http.Response {
+	a, err := r.join(ctx, n)
+	if err != nil {
+		r.logf("hosted: joining room %d: %v", n, err)
+		return xmlrpcFault("T5DE's hub isn't reachable right now.")
+	}
+	if a.Type != "room.joined" {
+		msg := a.Message
+		if msg == "" {
+			msg = "You can't join this T5DE room."
+		}
+		return xmlrpcFault(msg)
+	}
+	me := r.user()
+	taken := map[int]bool{}
+	for _, p := range a.Participants {
+		if p.ImvuID != me {
+			taken[p.Seat] = true
+		}
+	}
+	seat := 1
+	for taken[seat] {
+		seat++
+	}
+	r.mu.Lock()
+	r.last = n
+	_, known := r.owners[n]
+	r.mu.Unlock()
+	if !known {
+		r.getRoom(ctx, n) // its owner, for the room's products
+	}
+	if r.OnJoin != nil {
+		r.OnJoin(strconv.FormatInt(a.Chat, 10), instanceOf(n))
+	}
+	r.logf("hosted: in room %d (chat %d)", n, a.Chat)
+	return xmlrpcAnswer(xmlrpc.Struct{
+		{Name: "waiting", Value: int64(0)}, {Name: "lastMessageId", Value: int64(a.Sequence)},
+		{Name: "chatId", Value: a.Chat}, {Name: "partnerId", Value: int64(0)}, {Name: "seat", Value: int64(seat)},
+		{Name: "participantUrl", Value: fmt.Sprintf("https://api.imvu.com/chat/chat-%d/participants/user-%d", a.Chat, me)},
+	})
+}
+
+// others is the room's other people, as chat.getParticipants lists them.
+func (r *Rooms) others(n int64) []xmlrpc.Value {
+	me := r.user()
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := []xmlrpc.Value{}
+	if rm, ok := r.rooms[n]; ok {
+		for _, p := range rm.participants {
+			if p.ImvuID != me {
+				out = append(out, xmlrpc.Struct{{Name: "userId", Value: p.ImvuID}, {Name: "seat", Value: int64(p.Seat)}})
+			}
+		}
+	}
+	return out
+}
+
+// addToRoomList adds hosted rooms to the client's room list (rooms_list_paginated.php: one array per field): all
+// of them to the first page without a search, and those the client was in this run to recently visited rooms.
+func (r *Rooms) addToRoomList(x *proxy.Exchange) {
+	q := x.URL.Query()
+	if p := q.Get("page"); p != "" && p != "1" && p != "0" {
+		return
+	}
+	if q.Get("keywords") != "" || q.Get("search") != "" {
+		return
+	}
+	recent := q.Get("recent") != ""
+	x.ModifyResponse = func(resp *http.Response, body []byte) ([]byte, error) {
+		var list map[string]json.RawMessage
+		if resp.StatusCode != http.StatusOK || json.Unmarshal(body, &list) != nil {
+			return body, nil
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		var out struct {
+			Rooms []Room `json:"rooms"`
+		}
+		if err := r.Hub.Call(ctx, "GET", "/api/v1/hosted-rooms", nil, &out); err != nil || len(out.Rooms) == 0 {
+			return body, nil
+		}
+		sort.SliceStable(out.Rooms, func(i, j int) bool { return out.Rooms[i].Occupancy > out.Rooms[j].Occupancy })
+		r.mu.Lock()
+		visited := map[int64]bool{}
+		for n := range r.owners {
+			visited[n] = true
+		}
+		r.mu.Unlock()
+		for _, rm := range out.Rooms {
+			if !recent || visited[rm.N] {
+				appendRoom(list, rm)
+			}
+		}
+		return json.Marshal(list)
+	}
+}
+
+// appendRoom adds a room to each of the list's arrays (fields it doesn't know get an empty value of the array's
+// kind).
+func appendRoom(list map[string]json.RawMessage, rm Room) {
+	adult := "0"
+	if rm.Rating == "adult" {
+		adult = "1"
+	}
+	known := map[string]any{
+		"customers_room_id": strconv.FormatInt(rm.N, 10), "customers_id": strconv.Itoa(Base),
+		"room_instance_id": instanceOf(rm.N), "name": plain(rm.Name), "customers_name": plain(rm.OwnerName),
+		"num_participants": strconv.Itoa(rm.Occupancy), "max_users": strconv.Itoa(rm.MaxUsers),
+		"description": plain(rm.Description), "image_url": "", "rating": "0", "language": "", "is_ap": adult,
+		"is_av": "0", "is_vip": "0", "participants": []any{},
+	}
+	for name, raw := range list {
+		var arr []json.RawMessage
+		if json.Unmarshal(raw, &arr) != nil {
+			continue
+		}
+		v, ok := known[name]
+		if !ok {
+			v = emptyLike(arr)
+		} else if str, isString := v.(string); isString && len(arr) > 0 && !strings.HasPrefix(strings.TrimSpace(string(arr[0])), "\"") {
+			// The list has numbers here, not strings.
+			if n, err := strconv.ParseInt(str, 10, 64); err == nil {
+				v = n
+			}
+		}
+		b, _ := json.Marshal(v)
+		arr = append(arr, b)
+		list[name], _ = json.Marshal(arr)
+	}
+}
+
+// emptyLike is an empty value of the kind the array holds.
+func emptyLike(arr []json.RawMessage) any {
+	if len(arr) == 0 {
+		return ""
+	}
+	switch b := strings.TrimSpace(string(arr[0])); {
+	case strings.HasPrefix(b, "["):
+		return []any{}
+	case strings.HasPrefix(b, "{"):
+		return map[string]any{}
+	case strings.HasPrefix(b, "\""):
+		return ""
+	case b == "true" || b == "false":
+		return false
+	case b == "null":
+		return nil
+	}
+	return 0
+}
+
+// plain keeps text to what the client's pages can show however they insert it (as in the products package).
+func plain(s string) string {
+	return strings.Map(func(c rune) rune {
+		switch c {
+		case '<', '>', '&', '"', '\'', '`':
+			return -1
+		}
+		return c
+	}, s)
+}

internal/hosted/hosted_test.go

@@ -0,0 +1,259 @@
+package hosted
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"sync"
+	"testing"
+
+	"git.t5de.dev/t5de/t5de-proxy/internal/imq"
+	"git.t5de.dev/t5de/t5de-proxy/internal/proxy"
+	"git.t5de.dev/t5de/t5de-proxy/internal/xmlrpc"
+)
+
+// fakeHub answers joins at once (through rooms.FromHub) and records what it's sent.
+type fakeHub struct {
+	mu    sync.Mutex
+	rooms *Rooms
+	sent  []map[string]any
+	scene []any
+}
+
+func (h *fakeHub) Send(v any) bool {
+	m := v.(map[string]any)
+	h.mu.Lock()
+	h.sent = append(h.sent, m)
+	h.mu.Unlock()
+	if m["type"] == "room.join" {
+		n := m["room"].(int64)
+		var answer string
+		if n == 9 {
+			answer = `{"type": "room.error", "room": 9, "reason": "full", "message": "This room is full."}`
+		} else {
+			answer = fmt.Sprintf(`{"type": "room.joined", "room": %d, "chat": %d, "sequence": 4,
+				"participants": [{"imvu_id": 7, "seat": 1}, {"imvu_id": 88, "seat": 2}]}`, n, Base+n)
+		}
+		go h.rooms.FromHub([]byte(answer))
+	}
+	return true
+}
+
+func (h *fakeHub) Call(ctx context.Context, method, path string, body, out any) error {
+	switch {
+	case method == "GET" && path == "/api/v1/hosted-rooms/3":
+		return json.Unmarshal([]byte(`{"room": {"n": 3, "name": "Lounge <b>", "ownerImvuId": 88, "ownerName": "Erin",
+			"roomPid": 50168116, "rating": "general", "maxUsers": 10, "occupancy": 2},
+			"scene": {"contents": [[1, 47477426, "37207-46"], [2, "coatcoat22", "37-11"], [3, "notherenn2", "37-12"]], "hash": "h1"}}`), out)
+	case method == "GET" && path == "/api/v1/hosted-rooms":
+		return json.Unmarshal([]byte(`{"rooms": [{"n": 3, "name": "Lounge", "ownerName": "Erin", "maxUsers": 10, "occupancy": 2}]}`), out)
+	case method == "PUT" && path == "/api/v1/hosted-rooms/3/scene":
+		h.mu.Lock()
+		h.scene = append(h.scene, body)
+		h.mu.Unlock()
+		return nil
+	}
+	return fmt.Errorf("unexpected %s %s", method, path)
+}
+
+func (h *fakeHub) sentOf(kind string) []map[string]any {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	var out []map[string]any
+	for _, m := range h.sent {
+		if m["type"] == kind {
+			out = append(out, m)
+		}
+	}
+	return out
+}
+
+func setup() (*Rooms, *fakeHub, *[]string) {
+	h := &fakeHub{}
+	var events []string
+	r := &Rooms{Hub: h, User: func() int64 { return 7 },
+		OnJoin:        func(chat, room string) { events = append(events, "join "+chat+" "+room) },
+		OnLeave:       func(chat string) { events = append(events, "leave "+chat) },
+		LocalProduct:  func(id string) (int64, bool) { return 2000000005, id == "coatcoat22" },
+		GlobalProduct: func(local int64) (string, bool) { return "coatcoat22", local == 2000000005 }}
+	h.rooms = r
+	return r, h, &events
+}
+
+func xmlrpcCall(t *testing.T, r *Rooms, host, method string, args xmlrpc.Struct) (*xmlrpc.Response, bool) {
+	t.Helper()
+	body := (&xmlrpc.Call{Method: method, Params: []xmlrpc.Value{args}}).Marshal()
+	u, _ := url.Parse("https://" + host + "/api/xmlrpc/" + map[bool]string{true: "chat.php", false: "client.php"}[strings.HasPrefix(method, "chat.")])
+	req, _ := http.NewRequest("POST", u.String(), bytes.NewReader(body))
+	x := &proxy.Exchange{URL: u, Request: req, Body: body}
+	r.Intercept(x)
+	if x.Response == nil {
+		return nil, false
+	}
+	b, _ := io.ReadAll(x.Response.Body)
+	resp, err := xmlrpc.ParseResponse(b)
+	if err != nil {
+		t.Fatalf("%s: %v\n%s", method, err, b)
+	}
+	return resp, true
+}
+
+func get(v xmlrpc.Value, name string) xmlrpc.Value {
+	s, _ := v.(xmlrpc.Struct)
+	out, _ := s.Get(name)
+	return out
+}
+
+// frames reads the frames the proxy sent the client.
+func frames(t *testing.T, b []byte) []imq.Frame {
+	var out []imq.Frame
+	for len(b) > 0 {
+		f, n, err := imq.Next(b)
+		if err != nil || n == 0 {
+			t.Fatalf("frames: %v", err)
+		}
+		out = append(out, f)
+		b = b[n:]
+	}
+	return out
+}
+
+func TestHostedRoom(t *testing.T) {
+	r, h, events := setup()
+
+	// IMVU's rooms are left alone.
+	if _, answered := xmlrpcCall(t, r, "chat.imvu.com", "chat.getOrMakeChat", xmlrpc.Struct{{Name: "activity", Value: "publicroom-170879926-5533"}}); answered {
+		t.Fatal("answered an IMVU room")
+	}
+
+	// Joining: the hub lets the client in; it sits in the first free seat.
+	resp, _ := xmlrpcCall(t, r, "chat.imvu.com", "chat.getOrMakeChat", xmlrpc.Struct{{Name: "activity", Value: "publicroom-2100000000-3"}})
+	if get(resp.Params[0], "chatId") != int64(2100000003) || get(resp.Params[0], "seat") != int64(1) || get(resp.Params[0], "lastMessageId") != int64(4) {
+		t.Fatalf("join answer %v", resp.Params[0])
+	}
+	if len(*events) != 1 || (*events)[0] != "join 2100000003 2100000000-3" {
+		t.Fatalf("events %v", *events)
+	}
+	// A full room: the client is told why.
+	if resp, _ := xmlrpcCall(t, r, "chat.imvu.com", "chat.getOrMakeChat", xmlrpc.Struct{{Name: "activity", Value: "publicroom-2100000000-9"}}); resp.Fault == nil || resp.Fault.String != "This room is full." {
+		t.Fatalf("full room: %+v", resp)
+	}
+
+	// Others in the room, without the client.
+	resp, _ = xmlrpcCall(t, r, "chat.imvu.com", "chat.getParticipants", xmlrpc.Struct{{Name: "userId", Value: int64(7)}, {Name: "chatId", Value: int64(2100000003)}})
+	others := resp.Params[0].([]xmlrpc.Value)
+	if len(others) != 1 || get(others[0], "userId") != int64(88) {
+		t.Fatalf("participants %v", others)
+	}
+
+	// IMQ: the room's queues are answered here; IMVU gets the rest.
+	var toClient bytes.Buffer
+	c := proxy.NewIMQConn(1, &toClient)
+	sub := imq.Subscription{Queues: []string{"/user/7"}, WithResults: []imq.QueueOp{{Queue: "/chat/2100000003", OpID: 5}, {Queue: "/chat/1385863547", OpID: 6}}}
+	rest := r.Route(c, false, sub)
+	if fmt.Sprint(rest.All()) != "[/user/7 /chat/1385863547]" {
+		t.Fatalf("for IMVU: %v", rest.All())
+	}
+	var types []string
+	for _, f := range frames(t, toClient.Bytes()) {
+		types = append(types, imq.TypeName(f.Type))
+	}
+	if got := strings.Join(types, " "); got != "G2cJoinedQueue G2cCreateMount G2cCreateMount G2cCreateMount G2cResult" {
+		t.Fatalf("frames %s", got)
+	}
+
+	// Chat goes to the hub; sitting down too.
+	if !r.Send("/chat/2100000003", "messages", []byte(`{"message": "*seat 2", "userId": 7}`)) || r.Send("/chat/1385863547", "messages", []byte(`{}`)) {
+		t.Fatal("Send's answer")
+	}
+	if seats, sends := h.sentOf("room.seat"), h.sentOf("room.send"); len(seats) != 1 || seats[0]["seat"] != 2 || len(sends) != 1 {
+		t.Fatalf("seats %v, sends %v", seats, sends)
+	}
+
+	// Others' chat comes back as the gateway would deliver it.
+	toClient.Reset()
+	r.FromHub([]byte(`{"type": "room.message", "room": 3, "from": 88, "mount": "messages", "message": "{\"message\":\"hi\",\"userId\":88}", "sequence": 5}`))
+	fs := frames(t, toClient.Bytes())
+	if len(fs) != 1 {
+		t.Fatalf("%d frames", len(fs))
+	}
+	d, _ := imq.ParseDelivered(fs[0].Data)
+	if d.UserID != "/user/88" || d.Queue != "/chat/2100000003" || d.Sequence != 5 || !strings.Contains(string(d.Message), "hi") {
+		t.Fatalf("delivered %+v", d)
+	}
+
+	// The room's card and scene come from the hub; the scene's community products as local IDs (missing ones left out).
+	resp, _ = xmlrpcCall(t, r, "client-dynamic.imvu.com", "test.getRoomCardInfo", xmlrpc.Struct{{Name: "room_instance_id", Value: "2100000000-3"}})
+	if get(resp.Params[0], "name") != "Lounge b" || get(resp.Params[0], "owner_cid") != int64(88) {
+		t.Fatalf("card %v", resp.Params[0])
+	}
+	resp, _ = xmlrpcCall(t, r, "client-dynamic.imvu.com", "test.getRoomState", xmlrpc.Struct{{Name: "room_instance_id", Value: "2100000000-3"}})
+	contents := get(get(resp.Params[0], "room_state"), "room_contents").([]xmlrpc.Value)
+	if len(contents) != 2 || fmt.Sprint(contents[1]) != "[2 2000000005 37-11]" || get(get(get(resp.Params[0], "room_state"), "room_info"), "room_pid") != "50168116" {
+		t.Fatalf("scene %v", resp.Params[0])
+	}
+	// Saving the scene: community products as the hub's IDs.
+	xmlrpcCall(t, r, "client-dynamic.imvu.com", "test.setRoomState", xmlrpc.Struct{{Name: "room_instance_id", Value: "2100000000-3"},
+		{Name: "room_contents", Value: []xmlrpc.Value{[]xmlrpc.Value{int64(1), int64(47477426), "37-1"}, []xmlrpc.Value{int64(2), int64(2000000005), "37-2"}}}})
+	if b, _ := json.Marshal(h.scene); string(b) != `[{"contents":[[1,47477426,"37-1"],[2,"coatcoat22","37-2"]]}]` {
+		t.Fatalf("saved %s", b)
+	}
+
+	// The participant post for a hosted chat is answered here.
+	u, _ := url.Parse("https://api.imvu.com/chat/chat-2100000003/participants/user-7")
+	req, _ := http.NewRequest("POST", u.String(), nil)
+	x := &proxy.Exchange{URL: u, Request: req}
+	r.Intercept(x)
+	if x.Response == nil || x.Response.StatusCode != 200 {
+		t.Fatal("participant post not answered")
+	}
+
+	// Anything else naming a hosted room is answered here, not sent to IMVU.
+	for _, raw := range []string{"https://api.imvu.com/room/room-2100000000-3", "https://client-dynamic.imvu.com/api/rooms/room_rating.php?room_instance_id=2100000000-3",
+		"https://client-dynamic.imvu.com/api/whatever.php?chat=2100000003"} {
+		u, _ := url.Parse(raw)
+		req, _ := http.NewRequest("GET", u.String(), nil)
+		x := &proxy.Exchange{URL: u, Request: req}
+		r.Intercept(x)
+		if x.Response == nil {
+			t.Errorf("%s went to IMVU", raw)
+		}
+	}
+
+	// Leaving.
+	xmlrpcCall(t, r, "chat.imvu.com", "chat.terminateChat", xmlrpc.Struct{{Name: "userId", Value: int64(7)}, {Name: "chatId", Value: int64(2100000003)}})
+	if len(h.sentOf("room.leave")) != 1 || (*events)[len(*events)-1] != "leave 2100000003" {
+		t.Fatalf("leave: %v %v", h.sentOf("room.leave"), *events)
+	}
+}
+
+func TestRoomList(t *testing.T) {
+	r, _, _ := setup()
+	u, _ := url.Parse("https://client-dynamic.imvu.com/api/rooms/rooms_list_paginated.php?page=1")
+	req, _ := http.NewRequest("GET", u.String(), nil)
+	x := &proxy.Exchange{URL: u, Request: req}
+	r.Intercept(x)
+	body := []byte(`{"customers_room_id": ["644"], "customers_id": ["1"], "num_participants": [4], "name": ["A"], "participants": [[]], "is_ap": ["0"], "odd": [{"x": 1}], "number_of_rooms": 1}`)
+	out, err := x.ModifyResponse(&http.Response{StatusCode: 200}, body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var l map[string]any
+	json.Unmarshal(out, &l)
+	if fmt.Sprint(l["customers_room_id"]) != "[644 3]" || fmt.Sprint(l["customers_id"]) != "[1 2100000000]" ||
+		fmt.Sprint(l["num_participants"]) != "[4 2]" || fmt.Sprint(l["odd"]) != "[map[x:1] map[]]" || l["number_of_rooms"] != float64(1) {
+		t.Fatalf("list %s", out)
+	}
+	// Other pages and searches are left alone.
+	u, _ = url.Parse("https://client-dynamic.imvu.com/api/rooms/rooms_list_paginated.php?page=2")
+	x = &proxy.Exchange{URL: u, Request: req}
+	r.Intercept(x)
+	if x.ModifyResponse != nil {
+		t.Fatal("changed page 2")
+	}
+}

internal/hub/hub.go

@@ -176,6 +176,10 @@ type Client struct {
 	OnChat func(chat string, avatars []Avatar)
 	// OnLibrary, if set, is called when connected and whenever the person's library changes.
 	OnLibrary func()
+	// OnConnect, if set, is called each time the connection to the hub is made (to send state again).
+	OnConnect func()
+	// OnRoom, if set, gets the hub's messages about hosted rooms ("room.*"), as sent.
+	OnRoom func(data []byte)
 
 	mu     sync.Mutex
 	saved  *Saved
@@ -183,11 +187,13 @@ type Client struct {
 	outfit []Item
 	chats  map[string]string // chat ID -> room instance ("" for other chats)
 	wake   chan struct{}
+	outbox chan []byte // messages sent as they come (Send), not state
 }
 
 // New makes a client for the sign-in saved at path.
 func New(path string, saved *Saved, logger *log.Logger) *Client {
-	return &Client{path: path, saved: saved, log: logger, chats: map[string]string{}, wake: make(chan struct{}, 1)}
+	return &Client{path: path, saved: saved, log: logger, chats: map[string]string{}, wake: make(chan struct{}, 1),
+		outbox: make(chan []byte, 256)}
 }
 
 // Join reports the client is in an IMVU chat (room: the room instance, for a room's chat).
@@ -241,6 +247,22 @@ func (c *Client) Outfit(imvuID int64, items []Item) {
 	c.poke()
 }
 
+// Send sends a message to the hub once (hosted rooms' messages); false if it can't be queued. Messages queued while
+// disconnected are dropped on connecting (OnConnect sends what's still true).
+func (c *Client) Send(v any) bool {
+	b, err := json.Marshal(v)
+	if err != nil {
+		return false
+	}
+	select {
+	case c.outbox <- b:
+		return true
+	default:
+		c.log.Printf("hub: too many messages waiting; dropped one")
+		return false
+	}
+}
+
 func (c *Client) poke() {
 	select {
 	case c.wake <- struct{}{}:
@@ -289,6 +311,13 @@ func (c *Client) connect(ctx context.Context) error {
 	if c.OnLibrary != nil {
 		go c.OnLibrary()
 	}
+	for drained := false; !drained; {
+		select {
+		case <-c.outbox:
+		default:
+			drained = true
+		}
+	}
 
 	errs := make(chan error, 1)
 	go func() {
@@ -302,6 +331,17 @@ func (c *Client) connect(ctx context.Context) error {
 				errs <- err
 				return
 			}
+			var kind struct{ Type string }
+			if json.Unmarshal(data, &kind) != nil {
+				continue
+			}
+			if strings.HasPrefix(kind.Type, "room.") {
+				// Hosted rooms' messages, whose fields differ (chat IDs are numbers).
+				if c.OnRoom != nil {
+					c.OnRoom(data)
+				}
+				continue
+			}
 			var m struct {
 				Type, Message, Chat string
 				Avatars             []Avatar
@@ -349,6 +389,7 @@ func (c *Client) connect(ctx context.Context) error {
 	}()
 	sentID, sentOutfit := int64(-1), ""
 	sentChats := map[string]string{}
+	connected := false
 	for {
 		c.mu.Lock()
 		id, outfit := c.imvuID, append([]Item{}, c.outfit...) // a list, never null
@@ -396,12 +437,26 @@ func (c *Client) connect(ctx context.Context) error {
 				delete(sentChats, chat)
 			}
 		}
+		if !connected {
+			// After the session (so the hub knows the IMVU account), say what's still true.
+			connected = true
+			if c.OnConnect != nil {
+				go c.OnConnect()
+			}
+		}
 		select {
 		case <-ctx.Done():
 			conn.Close(websocket.StatusNormalClosure, "")
 			return ctx.Err()
 		case err := <-errs:
 			return err
+		case b := <-c.outbox:
+			wctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+			err := conn.Write(wctx, websocket.MessageText, b)
+			cancel()
+			if err != nil {
+				return err
+			}
 		case <-c.wake:
 		}
 	}

internal/imq/frame_test.go

@@ -81,3 +81,19 @@ func TestSendMessageRoundTrip(t *testing.T) {
 		t.Fatal("a G2cResult parsed as a C2gSendMessage")
 	}
 }
+
+func TestSubscriptionOps(t *testing.T) {
+	// A subscription with a plain queue, and one with an op ID and a property (field 3), which survives re-encoding.
+	withProps := appendBytesField(appendVarintField(appendBytesField(nil, 1, []byte("/chat/9")), 2, 7), 3, []byte("x"))
+	data := appendBytesField(appendBytesField(nil, 1, []byte("/user/1")), 2, withProps)
+	s, err := ParseSubscriptionOps(data)
+	if err != nil || len(s.Queues) != 1 || len(s.WithResults) != 1 || s.WithResults[0].Queue != "/chat/9" || s.WithResults[0].OpID != 7 {
+		t.Fatalf("%+v %v", s, err)
+	}
+	if string(s.Marshal()) != string(data) {
+		t.Fatal("re-encoded differently")
+	}
+	if all := s.All(); len(all) != 2 || all[1] != "/chat/9" {
+		t.Fatalf("all %v", all)
+	}
+}

internal/imq/messages.go

@@ -9,6 +9,16 @@ const (
 	C2gSubscribe   = 1309
 	G2cResult      = 3102
 	G2cSendMessage = 3104
+	G2cStateChange = 3105
+	G2cJoinedQueue = 3106
+	G2cLeftQueue   = 3107
+	G2cCreateMount = 3108
+)
+
+// Mount types (G2cCreateMount).
+const (
+	MessageMount = 1
+	StateMount   = 2
 )
 
 // SendMessage is a C2gSendMessage: a message the client sends to a queue's mount.
@@ -202,3 +212,105 @@ func (m Delivered) Marshal() []byte {
 	}
 	return b
 }
+
+// QueueOp is a queue subscribed to with an operation ID, for a G2cResult.
+type QueueOp struct {
+	Queue string
+	OpID  uint64
+	raw   []byte // as the client sent it (with any other fields, such as properties)
+}
+
+// Subscription is a C2gSubscribe's or C2gUnsubscribe's data: queues listed plainly, and with operation IDs.
+type Subscription struct {
+	Queues      []string
+	WithResults []QueueOp
+}
+
+// ParseSubscriptionOps reads a C2gSubscribe's or C2gUnsubscribe's data, keeping operation IDs.
+func ParseSubscriptionOps(data []byte) (Subscription, error) {
+	fs, err := fields(data)
+	if err != nil {
+		return Subscription{}, err
+	}
+	var s Subscription
+	for _, f := range fs {
+		switch {
+		case f.num == 1 && f.isLen:
+			s.Queues = append(s.Queues, string(f.bytes))
+		case f.num == 2 && f.isLen:
+			sub, err := fields(f.bytes)
+			if err != nil {
+				return Subscription{}, err
+			}
+			q := QueueOp{raw: f.bytes}
+			for _, sf := range sub {
+				switch {
+				case sf.num == 1 && sf.isLen:
+					q.Queue = string(sf.bytes)
+				case sf.num == 2 && !sf.isLen:
+					q.OpID = sf.varint
+				}
+			}
+			s.WithResults = append(s.WithResults, q)
+		default:
+			return Subscription{}, errors.New("imq: unexpected field in a subscription")
+		}
+	}
+	return s, nil
+}
+
+// All is every queue in the subscription.
+func (s Subscription) All() []string {
+	out := append([]string(nil), s.Queues...)
+	for _, q := range s.WithResults {
+		out = append(out, q.Queue)
+	}
+	return out
+}
+
+// Empty reports whether it names no queues.
+func (s Subscription) Empty() bool { return len(s.Queues) == 0 && len(s.WithResults) == 0 }
+
+// Marshal encodes the subscription's data.
+func (s Subscription) Marshal() []byte {
+	var b []byte
+	for _, q := range s.Queues {
+		b = appendBytesField(b, 1, []byte(q))
+	}
+	for _, q := range s.WithResults {
+		raw := q.raw
+		if raw == nil {
+			raw = appendVarintField(appendBytesField(nil, 1, []byte(q.Queue)), 2, q.OpID)
+		}
+		b = appendBytesField(b, 2, raw)
+	}
+	return b
+}
+
+// Property is a mount's property: a key and a value.
+type Property struct{ Key, Value string }
+
+func appendProperties(b []byte, num uint64, props []Property) []byte {
+	for _, p := range props {
+		b = appendBytesField(b, num, appendBytesField(appendBytesField(nil, 1, []byte(p.Key)), 2, []byte(p.Value)))
+	}
+	return b
+}
+
+// JoinedQueue encodes a G2cJoinedQueue's (or G2cLeftQueue's) data: a user joined (left) a queue.
+func JoinedQueue(userID, queue string, sequence uint64) []byte {
+	return appendVarintField(appendBytesField(appendBytesField(nil, 1, []byte(userID)), 2, []byte(queue)), 3, sequence)
+}
+
+// CreateMount encodes a G2cCreateMount's data: a queue's mount, of type MessageMount or StateMount.
+func CreateMount(queue, mount string, mountType uint64, props []Property, sequence uint64) []byte {
+	b := appendVarintField(appendBytesField(appendBytesField(nil, 1, []byte(queue)), 2, []byte(mount)), 3, mountType)
+	return appendVarintField(appendProperties(b, 4, props), 5, sequence)
+}
+
+// StateChange encodes a G2cStateChange's data: a state mount's properties changed (an empty value removes one).
+func StateChange(opID uint64, userID, queue, mount string, props []Property, sequence uint64) []byte {
+	b := appendVarintField(nil, 1, opID)
+	b = appendBytesField(appendBytesField(appendBytesField(b, 2, []byte(userID)), 3, []byte(queue)), 4, []byte(mount))
+	return appendVarintField(appendProperties(b, 5, props), 6, sequence)
+}

internal/products/intercept.go

@@ -359,6 +359,13 @@ func (p *Products) authorizeProducts2(x *proxy.Exchange, call *xmlrpc.Call) erro
 	type auth struct {
 		user, product int64
 	}
+	// In a room hosted by T5DE's hub, the client asks about the room's products as user 2100000000 (the first part
+	// of the room's ID): IMVU is asked about the room owner's instead, and answers go back as 2100000000.
+	owner := int64(0)
+	if p.HostedOwner != nil {
+		owner = p.HostedOwner()
+	}
+	hostedAsked := map[int64]bool{}
 	var community []auth
 	forward := []xmlrpc.Value{table[0]}
 	asked := map[auth]bool{}
@@ -372,6 +379,10 @@ func (p *Products) authorizeProducts2(x *proxy.Exchange, call *xmlrpc.Call) erro
 		if !ok1 || !ok2 {
 			return nil
 		}
+		if user == HostedOwnerID && owner > 0 {
+			user, hostedAsked[product] = owner, true
+			row = []xmlrpc.Value{user, product}
+		}
 		if !IsCommunity(product) {
 			if !asked[auth{user, product}] {
 				forward, asked[auth{user, product}] = append(forward, row), true
@@ -386,9 +397,33 @@ func (p *Products) authorizeProducts2(x *proxy.Exchange, call *xmlrpc.Call) erro
 			}
 		}
 	}
-	if len(community) == 0 {
+	if len(community) == 0 && len(hostedAsked) == 0 {
 		return nil
 	}
+	// hostedBack gives rows asked about as 2100000000 that user ID back.
+	hostedBack := func(columns []xmlrpc.Value, rows []xmlrpc.Value) {
+		userCol, productCol := -1, -1
+		for i, c := range columns {
+			switch c {
+			case "userId":
+				userCol = i
+			case "productId":
+				productCol = i
+			}
+		}
+		if userCol < 0 || productCol < 0 {
+			return
+		}
+		for _, row := range rows {
+			r, ok := row.([]xmlrpc.Value)
+			if !ok || len(r) <= userCol || len(r) <= productCol {
+				continue
+			}
+			if product, _ := r[productCol].(int64); r[userCol] == owner && hostedAsked[product] {
+				r[userCol] = int64(HostedOwnerID)
+			}
+		}
+	}
 	ours := func(columns []xmlrpc.Value) []xmlrpc.Value {
 		var rows []xmlrpc.Value
 		for _, a := range community {
@@ -421,6 +456,7 @@ func (p *Products) authorizeProducts2(x *proxy.Exchange, call *xmlrpc.Call) erro
 			columns[i] = c
 		}
 		out := append([]xmlrpc.Value{columns}, ours(columns)...)
+		hostedBack(columns, out[1:])
 		x.Response = localResponse(200, "text/xml", (&xmlrpc.Response{Params: []xmlrpc.Value{out}}).Marshal())
 		return nil
 	}
@@ -447,7 +483,9 @@ func (p *Products) authorizeProducts2(x *proxy.Exchange, call *xmlrpc.Call) erro
 		if !ok {
 			return body, errors.New("products: unexpected authorizeProducts2 columns")
 		}
-		r.Params[0] = append(rows, ours(columns)...)
+		all := append(rows, ours(columns)...)
+		hostedBack(columns, all[1:])
+		r.Params[0] = all
 		return r.Marshal(), nil
 	}
 	return nil
@@ -523,6 +561,15 @@ func (p *Products) interceptBaseLook(x *proxy.Exchange) error {
 // Outgoing implements proxy.IMQFilter: community products are taken out of outfit chat commands (IMVU and other
 // people see the defaults), and whole outfits are remembered.
 func (p *Products) Outgoing(queue, mount string, message []byte) ([]byte, bool) {
+	out, drop := p.outgoing(queue, mount, message)
+	// A hosted room's message goes to the hub (without community products, as to IMVU), never to IMVU.
+	if p.Hosted != nil && !drop && p.Hosted.Send(queue, mount, out) {
+		return nil, true
+	}
+	return out, drop
+}
+
+func (p *Products) outgoing(queue, mount string, message []byte) ([]byte, bool) {
 	if !strings.HasPrefix(queue, "/chat/") || mount != "messages" {
 		return message, false
 	}

internal/products/products.go

@@ -30,6 +30,8 @@ import (
 	"git.t5de.dev/t5de/t5de-api/cfl"
 	"git.t5de.dev/t5de/t5de-api/pack"
 	"git.t5de.dev/t5de/t5de-proxy/internal/hub"
+	"git.t5de.dev/t5de/t5de-proxy/internal/imq"
+	"git.t5de.dev/t5de/t5de-proxy/internal/proxy"
 )
 
 // Products is the proxy's community products: an Interceptor for HTTP and an IMQFilter for chat.
@@ -67,6 +69,11 @@ type Products struct {
 	// Library, if set, is the person's hub library as last read (for the inventory's Recently Added).
 	Library func() []hub.LibraryEntry
 
+	// Hosted, if set, serves T5DE-hosted rooms' queues (the hosted package); HostedOwner is the IMVU account owning
+	// the hosted room the client is in (0 if none), whose products the room's are.
+	Hosted      HostedRooms
+	HostedOwner func() int64
+
 	roomMu sync.Mutex
 	room   room
 
@@ -99,6 +106,26 @@ func (p *Products) Ready(id string) {
 	}
 }
 
+// HostedOwnerID is the user ID the client gives hosted rooms' owner: the first part of their instance IDs
+// (hosted.Base).
+const HostedOwnerID = 2100000000
+
+// HostedRooms serves T5DE-hosted rooms' IMQ queues: Route answers subscriptions to them, Send takes the client's
+// messages to them (reporting whether the queue was one), Closed forgets a connection.
+type HostedRooms interface {
+	Route(c *proxy.IMQConn, unsubscribe bool, s imq.Subscription) imq.Subscription
+	Send(queue, mount string, message []byte) bool
+	Closed(c *proxy.IMQConn)
+}
+
+// Route serves hosted rooms' queues (proxy.IMQRouter).
+func (p *Products) Route(c *proxy.IMQConn, unsubscribe bool, s imq.Subscription) imq.Subscription {
+	if p.Hosted == nil {
+		return s
+	}
+	return p.Hosted.Route(c, unsubscribe, s)
+}
+
 // Observer hears about the client's IMVU account, its community outfit, and the chats it's in.
 type Observer interface {
 	Session(imvuID int64)
@@ -371,3 +398,22 @@ func (p *Products) reportOutfit(user string, locals []int64) {
 	}
 	p.Observer.Outfit(id, items)
 }
+
+// LocalID is the client's local ID for a product (a hub product ID or a local pack's), if it's here.
+func (p *Products) LocalID(id string) (int64, bool) {
+	if s := p.byProductID(id); s != nil {
+		return s.local, true
+	}
+	return 0, false
+}
+
+// ProductID is the product a local ID was given to.
+func (p *Products) ProductID(local int64) (string, bool) { return p.state.productID(local) }
+
+// User is the IMVU account the client signed in as (0 before it has).
+func (p *Products) User() int64 {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	id, _ := strconv.ParseInt(p.user, 10, 64)
+	return id
+}

internal/products/products_test.go

@@ -417,3 +417,24 @@ func TestSavedOutfits(t *testing.T) {
 		t.Fatalf("outfits %s", body)
 	}
 }
+
+func TestHostedRoomProducts(t *testing.T) {
+	h, _ := newHarness(t)
+	h.do("POST", "https://secure.imvu.com/api/xmlrpc/client.php", call("test.avatarInfoForLogin2"), false, false)
+	h.p.HostedOwner = func() int64 { return 7 } // the fake IMVU answers as user 7
+	auths := xmlrpc.Struct{{Name: "auths", Value: []xmlrpc.Value{
+		[]xmlrpc.Value{"userId", "productId"}, []xmlrpc.Value{int64(HostedOwnerID), int64(500)}}}}
+	_, body := h.do("POST", "http://client-dynamic.imvu.com/api/xmlrpc/client.php", call("test.authorizeProducts2", auths), true, false)
+	asked := h.imvu.asked["test.authorizeProducts2"]
+	if len(asked) != 1 || strings.Contains(asked[0], "2100000000") {
+		t.Fatalf("IMVU was asked %q", asked)
+	}
+	r, err := xmlrpc.ParseResponse(body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rows := r.Params[0].([]xmlrpc.Value)
+	if first := rows[1].([]xmlrpc.Value); first[0] != int64(HostedOwnerID) || first[2] != int64(500) {
+		t.Fatalf("answer %v", rows)
+	}
+}

internal/products/room.go

@@ -116,6 +116,9 @@ func (p *Products) Unsubscribed(c *proxy.IMQConn, queues []string) {
 
 // Closed implements proxy.IMQWatcher: a connection's subscriptions end with it.
 func (p *Products) Closed(c *proxy.IMQConn) {
+	if p.Hosted != nil {
+		p.Hosted.Closed(c)
+	}
 	p.roomMu.Lock()
 	p.room.init()
 	var queues []string

internal/proxy/connect.go

@@ -123,6 +123,13 @@ type IMQWatcher interface {
 	Closed(c *IMQConn)
 }
 
+// IMQRouter is an IMQWatcher that serves some queues itself (T5DE-hosted rooms' queues): it takes them out of the
+// client's subscriptions before they go to the gateway, and answers the client for them.
+type IMQRouter interface {
+	// Route returns the part of a subscription (or unsubscription) for the gateway.
+	Route(c *IMQConn, unsubscribe bool, s imq.Subscription) imq.Subscription
+}
+
 // IMQConn is one IMQ connection of the client's.
 type IMQConn struct {
 	ID int64
@@ -132,6 +139,12 @@ type IMQConn struct {
 // NewIMQConn is a connection writing frames to w (for tests of watchers).
 func NewIMQConn(id int64, w io.Writer) *IMQConn { return &IMQConn{ID: id, w: w} }
 
+// Send gives the client a frame of the given type as if the gateway had sent it.
+func (c *IMQConn) Send(frameType uint64, data []byte) error {
+	_, err := c.w.Write(imq.EncodeFrame(frameType, data))
+	return err
+}
+
 // Deliver gives the client a message as if the gateway had delivered it.
 func (c *IMQConn) Deliver(m imq.Delivered) error {
 	_, err := c.w.Write(imq.EncodeFrame(imq.G2cSendMessage, m.Marshal()))
@@ -225,14 +238,27 @@ func (st *stream) filtered(f imq.Frame, frame []byte) []byte {
 	if st.watcher != nil {
 		switch f.Type {
 		case imq.C2gSubscribe, imq.C2gUnsubscribe:
-			if queues, err := imq.ParseSubscription(f.Data); err == nil {
+			sub, err := imq.ParseSubscriptionOps(f.Data)
-				if f.Type == imq.C2gSubscribe {
+			if err != nil {
-					st.watcher.Subscribed(st.conn, queues)
+				return frame
-				} else {
-					st.watcher.Unsubscribed(st.conn, queues)
-				}
 			}
-			return frame
+			if f.Type == imq.C2gSubscribe {
+				st.watcher.Subscribed(st.conn, sub.All())
+			} else {
+				st.watcher.Unsubscribed(st.conn, sub.All())
+			}
+			router, ok := st.watcher.(IMQRouter)
+			if !ok {
+				return frame
+			}
+			rest := router.Route(st.conn, f.Type == imq.C2gUnsubscribe, sub)
+			if rest.Empty() {
+				return nil
+			}
+			if len(rest.All()) == len(sub.All()) {
+				return frame
+			}
+			return imq.EncodeFrame(f.Type, rest.Marshal())
 		case imq.G2cSendMessage:
 			m, err := imq.ParseDelivered(f.Data)
 			if err != nil {