From 5b4e44c4e7ddfa64f35fe246f8a25af0518f7447 Mon Sep 17 00:00:00 2001 From: dhkatz Date: Thu, 29 Oct 2026 19:16:30 -0700 Subject: [PATCH] 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/. Checked live: removing a product from the library on the hub reloads the client's inventory. Co-Authored-By: Claude Opus 5.5 --- internal/products/intercept.go | 5 +++++ internal/products/products.go | 2 ++ internal/products/room.go | 34 +++++++++++++++++++++++----------- internal/proxy/connect.go | 3 +++ internal/proxy/proxy_test.go | 1 + 5 files changed, 34 insertions(+), 11 deletions(-) diff --git a/internal/products/intercept.go b/internal/products/intercept.go index a10859c..40f8a2a 100644 --- a/internal/products/intercept.go +++ b/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) diff --git a/internal/products/products.go b/internal/products/products.go index ca977d2..31df483 100644 --- a/internal/products/products.go +++ b/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 diff --git a/internal/products/room.go b/internal/products/room.go index a44359a..519bb41 100644 --- a/internal/products/room.go +++ b/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/") + 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/") 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)) - for c, q := range p.room.users { - users[c] = q + var conns []*proxy.IMQConn + for c := range p.room.conns { + 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) } diff --git a/internal/proxy/connect.go b/internal/proxy/connect.go index 584bf9c..d538e92 100644 --- a/internal/proxy/connect.go +++ b/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) diff --git a/internal/proxy/proxy_test.go b/internal/proxy/proxy_test.go index 0f9db71..aba029d 100644 --- a/internal/proxy/proxy_test.go +++ b/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)