Refresh the inventory on the user's queue, which the client joins by connecting

The client never subscribes to its user queue; the proxy now follows every IMQ connection and the signed-in
account (from the login answer), and delivers the refresh event on /user/<id>. Checked live: removing a product
from the library on the hub reloads the client's inventory.

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

internal/products/intercept.go

@@ -60,6 +60,11 @@ func (p *Products) interceptXMLRPC(x *proxy.Exchange) error {
 	case "test.avatarInfoForLogin2":
 		x.ModifyResponse = func(resp *http.Response, body []byte) ([]byte, error) {
 			p.keys.learn(body)
+			if m := userIDPattern.FindSubmatch(body); m != nil {
+				p.mu.Lock()
+				p.user = string(m[1])
+				p.mu.Unlock()
+			}
 			if m := userIDPattern.FindSubmatch(body); m != nil && p.Observer != nil {
 				if id, err := strconv.ParseInt(string(m[1]), 10, 64); err == nil {
 					p.Observer.Session(id)

internal/products/products.go

@@ -60,6 +60,8 @@ type Products struct {
 
 	roomMu sync.Mutex
 	room   room
+
+	user string // the signed-in IMVU account (from the login answer), under mu
 }
 
 // Ready is told a product arrived (the fetcher wrote it): it's served from now on, and given to the client for

internal/products/room.go

@@ -31,7 +31,7 @@ type room struct {
 	userIDs map[int64]string                   // user -> the gateway's user_id for them, as last seen
 	missing map[string]bool                    // products others wear that aren't in the packs (logged once)
 	last    map[string][]hub.Avatar            // chat -> what the hub last said about the others in it
-	users   map[*proxy.IMQConn]string          // each connection's user queue ("/user/<id>")
+	conns   map[*proxy.IMQConn]bool            // the client's IMQ connections
 }
 
 func (r *room) init() {
@@ -42,7 +42,7 @@ func (r *room) init() {
 		r.userIDs = map[int64]string{}
 		r.missing = map[string]bool{}
 		r.last = map[string][]hub.Avatar{}
-		r.users = map[*proxy.IMQConn]string{}
+		r.conns = map[*proxy.IMQConn]bool{}
 	}
 }
 
@@ -72,9 +72,6 @@ func (p *Products) Subscribed(c *proxy.IMQConn, queues []string) {
 	var joined []string
 	var rooms []string
 	for _, q := range queues {
-		if strings.HasPrefix(q, "/user/") {
-			p.room.users[c] = q
-		}
 		chat := chatOf(q)
 		if chat == "" {
 			continue
@@ -129,7 +126,7 @@ func (p *Products) Closed(c *proxy.IMQConn) {
 	p.Unsubscribed(c, queues)
 	p.roomMu.Lock()
 	delete(p.room.subs, c)
-	delete(p.room.users, c)
+	delete(p.room.conns, c)
 	p.roomMu.Unlock()
 }
 
@@ -365,17 +362,32 @@ func (p *Products) interceptParticipant(x *proxy.Exchange) error {
 	return nil
 }
 
+// Opened implements proxy.IMQWatcher.
+func (p *Products) Opened(c *proxy.IMQConn) {
+	p.roomMu.Lock()
+	defer p.roomMu.Unlock()
+	p.room.init()
+	p.room.conns[c] = true
+}
+
 // RefreshInventory has the client load its inventory again (after the library changed), with the server event
-// IMVU sends after a purchase: "refreshInventory {}" on the user queue's web_msg mount.
+// IMVU sends after a purchase: "refreshInventory {}" on the user queue's ("/user/<id>") web_msg mount.
 func (p *Products) RefreshInventory() {
+	p.mu.Lock()
+	user := p.user
+	p.mu.Unlock()
+	if user == "" {
+		return
+	}
 	p.roomMu.Lock()
 	p.room.init()
-	users := make(map[*proxy.IMQConn]string, len(p.room.users))
+	var conns []*proxy.IMQConn
-	for c, q := range p.room.users {
+	for c := range p.room.conns {
-		users[c] = q
+		conns = append(conns, c)
 	}
 	p.roomMu.Unlock()
-	for c, q := range users {
+	q := "/user/" + user
+	for _, c := range conns {
 		if err := c.Deliver(imq.Delivered{UserID: "0", Queue: q, Mount: "web_msg", Message: []byte("refreshInventory {}")}); err != nil {
 			p.logf("products: refreshing the inventory: %v", err)
 		}

internal/proxy/connect.go

@@ -74,6 +74,7 @@ func (s *Server) serveConnect(w http.ResponseWriter, r *http.Request) {
 		c := &IMQConn{ID: id, w: toClient}
 		down.watcher, down.conn = w, c
 		up.watcher, up.conn = w, c
+		w.Opened(c)
 		defer w.Closed(c)
 	}
 	// IMQ is a session: when either side closes, it's over, so both connections close.
@@ -111,6 +112,8 @@ type IMQFilter interface {
 // messages the gateway delivers (which it can change), and it can deliver messages of its own.
 type IMQWatcher interface {
 	IMQFilter
+	// Opened: the client opened an IMQ connection (its user queue comes with it, without a subscription).
+	Opened(c *IMQConn)
 	// Subscribed and Unsubscribed see the client's subscriptions on a connection.
 	Subscribed(c *IMQConn, queues []string)
 	Unsubscribed(c *IMQConn, queues []string)

internal/proxy/proxy_test.go

@@ -498,6 +498,7 @@ func (w *testWatcher) Incoming(c *IMQConn, m imq.Delivered) []byte {
 	return bytes.ToUpper(m.Message)
 }
 func (w *testWatcher) Closed(c *IMQConn) { w.add("closed") }
+func (w *testWatcher) Opened(c *IMQConn) {}
 
 func TestConnectWatchesSubscriptionsAndDeliveries(t *testing.T) {
 	gateway := httptest.NewUnstartedServer(nil)