t5de-proxy/internal/products/room.go

package products

// The room channel: other T5DE users' community products, on their avatars in this client.
//
// IMVU never sees community products, so other people's clients get each avatar's outfit without them. The hub
// tells the proxy which T5DE users are in the same chat and what they wear (by product ID); for products the proxy
// has, it delivers "*use" and "*remove" lines from those users to the client, as if they had sent them, and adds the
// products to the whole outfits those users send (which would otherwise put IMVU's products back in their place).
// The client then asks for the products like any other, and the proxy answers for them.

import (
	"bytes"
	"encoding/json"
	"net/http"
	"regexp"
	"slices"
	"strconv"
	"strings"

	"git.t5de.dev/t5de/t5de-proxy/internal/hub"
	"git.t5de.dev/t5de/t5de-proxy/internal/imq"
	"git.t5de.dev/t5de/t5de-proxy/internal/proxy"
	"git.t5de.dev/t5de/t5de-proxy/internal/xmlrpc"
)

// room is the room channel's state, under Products.roomMu.
type room struct {
	subs    map[*proxy.IMQConn]map[string]bool // chat IDs each IMQ connection is subscribed to
	rooms   map[string]string                  // chat ID -> room instance, from getOrMakeChat
	applied map[string]map[int64][]int64       // chat -> user -> local IDs the client was given for them
	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
	conns   map[*proxy.IMQConn]bool            // the client's IMQ connections
}

func (r *room) init() {
	if r.subs == nil {
		r.subs = map[*proxy.IMQConn]map[string]bool{}
		r.rooms = map[string]string{}
		r.applied = map[string]map[int64][]int64{}
		r.userIDs = map[int64]string{}
		r.missing = map[string]bool{}
		r.last = map[string][]hub.Avatar{}
		r.conns = map[*proxy.IMQConn]bool{}
	}
}

// chatOf is the chat ID of an IMQ chat queue ("/chat/<id>"), or "".
func chatOf(queue string) string {
	id, ok := strings.CutPrefix(queue, "/chat/")
	if !ok || id == "" || strings.Trim(id, "0123456789") != "" {
		return ""
	}
	return id
}

// inChat reports whether any connection is subscribed to chat (under roomMu).
func (r *room) inChat(chat string) bool {
	for _, chats := range r.subs {
		if chats[chat] {
			return true
		}
	}
	return false
}

// Subscribed implements proxy.IMQWatcher: chats the client joins are reported to the hub.
func (p *Products) Subscribed(c *proxy.IMQConn, queues []string) {
	p.roomMu.Lock()
	p.room.init()
	var joined []string
	var rooms []string
	for _, q := range queues {
		chat := chatOf(q)
		if chat == "" {
			continue
		}
		if !p.room.inChat(chat) {
			joined, rooms = append(joined, chat), append(rooms, p.room.rooms[chat])
		}
		if p.room.subs[c] == nil {
			p.room.subs[c] = map[string]bool{}
		}
		p.room.subs[c][chat] = true
	}
	p.roomMu.Unlock()
	if p.Observer != nil {
		for i, chat := range joined {
			p.Observer.Join(chat, rooms[i])
		}
	}
}

// Unsubscribed implements proxy.IMQWatcher.
func (p *Products) Unsubscribed(c *proxy.IMQConn, queues []string) {
	p.roomMu.Lock()
	p.room.init()
	var left []string
	for _, q := range queues {
		if chat := chatOf(q); chat != "" && p.room.subs[c][chat] {
			delete(p.room.subs[c], chat)
			if !p.room.inChat(chat) {
				left = append(left, chat)
				delete(p.room.applied, chat)
			}
		}
	}
	p.roomMu.Unlock()
	if p.Observer != nil {
		for _, chat := range left {
			p.Observer.Leave(chat)
		}
	}
}

// Closed implements proxy.IMQWatcher: a connection's subscriptions end with it.
func (p *Products) Closed(c *proxy.IMQConn) {
	p.roomMu.Lock()
	p.room.init()
	var queues []string
	for chat := range p.room.subs[c] {
		queues = append(queues, "/chat/"+chat)
	}
	p.roomMu.Unlock()
	p.Unsubscribed(c, queues)
	p.roomMu.Lock()
	delete(p.room.subs, c)
	delete(p.room.conns, c)
	p.roomMu.Unlock()
}

// userOf is the user ID in the gateway's user_id ("123" or ".../123").
func userOf(userID string) int64 {
	i := strings.LastIndexByte(userID, '/')
	id, err := strconv.ParseInt(userID[i+1:], 10, 64)
	if err != nil {
		return 0
	}
	return id
}

// Incoming implements proxy.IMQWatcher: a T5DE user's whole outfit gets their community products added.
func (p *Products) Incoming(c *proxy.IMQConn, m imq.Delivered) []byte {
	chat := chatOf(m.Queue)
	if chat == "" || m.Mount != "messages" {
		return m.Message
	}
	user := userOf(m.UserID)
	if user == 0 {
		return m.Message
	}
	p.roomMu.Lock()
	p.room.init()
	p.room.userIDs[user] = m.UserID
	locals := slices.Clone(p.room.applied[chat][user])
	p.roomMu.Unlock()
	if len(locals) == 0 {
		return m.Message
	}
	msg, ok := parseChatMessage(m.Message)
	if !ok {
		return m.Message
	}
	fields := strings.Fields(msg.text)
	if len(fields) < 2 || fields[0] != "*use" || !slices.Contains(fields[1:], strconv.Itoa(bodyProduct)) {
		return m.Message
	}
	for _, id := range locals {
		if s := strconv.FormatInt(id, 10); !slices.Contains(fields, s) {
			fields = append(fields, s)
		}
	}
	return msg.withText(strings.Join(fields, " "))
}

// Others is the hub's word on the other T5DE users in a chat and what they wear (none: forget them). The client is
// given the community products it has packs for, and told to take off the ones no longer worn.
func (p *Products) Others(chat string, avatars []hub.Avatar) {
	type line struct {
		user int64
		text string
	}
	want := map[int64][]int64{}
	for _, a := range avatars {
		var locals []int64
		for _, it := range a.Items {
			if s := p.byProductID(it.ID); s != nil {
				locals = append(locals, s.local)
				continue
			}
			if p.Want != nil {
				p.Want(it.ID) // given to the client when it arrives (Ready)
			}
			p.roomMu.Lock()
			p.room.init()
			first := !p.room.missing[it.ID]
			p.room.missing[it.ID] = true
			p.roomMu.Unlock()
			if first {
				p.logf("products: someone in the room wears %q (%s), which isn't here yet", it.Name, it.ID)
			}
		}
		slices.Sort(locals)
		want[a.IMVUID] = locals
	}

	p.roomMu.Lock()
	p.room.init()
	if avatars == nil {
		delete(p.room.applied, chat)
		delete(p.room.last, chat)
		p.roomMu.Unlock()
		return
	}
	p.room.last[chat] = avatars
	had := p.room.applied[chat]
	var lines []line
	for user, old := range had {
		if _, still := want[user]; !still && len(old) > 0 {
			lines = append(lines, line{user, "*remove " + joinIDs(old)})
		}
	}
	for user, locals := range want {
		old := had[user]
		var gone, added []int64
		for _, id := range old {
			if !slices.Contains(locals, id) {
				gone = append(gone, id)
			}
		}
		for _, id := range locals {
			if !slices.Contains(old, id) {
				added = append(added, id)
			}
		}
		if len(gone) > 0 {
			lines = append(lines, line{user, "*remove " + joinIDs(gone)})
		}
		if len(added) > 0 {
			lines = append(lines, line{user, "*use " + joinIDs(added)})
		}
	}
	p.room.applied[chat] = want
	var conns []*proxy.IMQConn
	for c, chats := range p.room.subs {
		if chats[chat] {
			conns = append(conns, c)
		}
	}
	userIDs := map[int64]string{}
	for _, l := range lines {
		if id, ok := p.room.userIDs[l.user]; ok {
			userIDs[l.user] = id
		} else {
			userIDs[l.user] = strconv.FormatInt(l.user, 10)
		}
	}
	p.roomMu.Unlock()

	chatID, _ := strconv.ParseInt(chat, 10, 64)
	for _, l := range lines {
		message, _ := json.Marshal(map[string]any{"userId": l.user, "chatId": chatID, "message": l.text, "to": 0})
		for _, c := range conns {
			if err := c.Deliver(imq.Delivered{UserID: userIDs[l.user], Queue: "/chat/" + chat, Mount: "messages", Message: message}); err != nil {
				p.logf("products: giving the client %d's community products: %v", l.user, err)
			}
		}
	}
	if len(lines) > 0 {
		p.logf("products: %d community product changes for others in chat %s", len(lines), chat)
	}
}

func joinIDs(ids []int64) string {
	s := make([]string, len(ids))
	for i, id := range ids {
		s[i] = strconv.FormatInt(id, 10)
	}
	return strings.Join(s, " ")
}

// interceptChat learns which room a chat is (chat.getOrMakeChat: asked with activity "publicroom-<room>",
// answered with the chat ID).
func (p *Products) interceptChat(x *proxy.Exchange) error {
	call, err := xmlrpc.ParseCall(x.Body)
	if err != nil || call.Method != "chat.getOrMakeChat" || len(call.Params) != 1 {
		return nil
	}
	args, ok := call.Params[0].(xmlrpc.Struct)
	if !ok {
		return nil
	}
	activity, _ := args.Get("activity")
	a, _ := activity.(string)
	inst, ok := strings.CutPrefix(a, "publicroom-")
	if !ok || !roomInstance.MatchString(inst) {
		return nil
	}
	x.ModifyResponse = func(resp *http.Response, body []byte) ([]byte, error) {
		r, err := xmlrpc.ParseResponse(body)
		if err != nil || r.Fault != nil || len(r.Params) != 1 {
			return body, nil
		}
		res, ok := r.Params[0].(xmlrpc.Struct)
		if !ok {
			return body, nil
		}
		v, _ := res.Get("chatId")
		var chat string
		switch v := v.(type) {
		case int64:
			chat = strconv.FormatInt(v, 10)
		case string:
			chat = v
		}
		if chat != "" && strings.Trim(chat, "0123456789") == "" {
			p.roomMu.Lock()
			p.room.init()
			p.room.rooms[chat] = inst
			p.roomMu.Unlock()
		}
		return body, nil
	}
	return nil
}

var roomInstance = regexp.MustCompile(`^[0-9]{1,12}-[0-9]{1,12}$`)

// participantPath is where the client posts its outfit and seat to IMVU while in a chat.
var participantPath = regexp.MustCompile(`^/chat/chat-[0-9]+/participants/user-[0-9]+$`)

// interceptParticipant keeps community products out of the outfit the client posts to IMVU's chat participant
// endpoint ({"legacy_outfit_message": "*use ...", ...}).
func (p *Products) interceptParticipant(x *proxy.Exchange) error {
	if len(x.Body) == 0 || !bytes.Contains(x.Body, []byte("legacy_outfit_message")) {
		return nil
	}
	var post map[string]json.RawMessage
	if json.Unmarshal(x.Body, &post) != nil {
		return nil
	}
	var text string
	if json.Unmarshal(post["legacy_outfit_message"], &text) != nil {
		return nil
	}
	r := stripCommand(text, IsCommunity)
	if !r.changed {
		return nil
	}
	if r.empty {
		delete(post, "legacy_outfit_message")
	} else {
		post["legacy_outfit_message"], _ = json.Marshal(r.text)
	}
	body, _ := json.Marshal(post)
	if !p.keys.resign(x.Request.Header, x.Body, x.URL.RawQuery, body, x.URL.RawQuery) {
		return errNoKey
	}
	p.logf("products: kept %d community products out of the outfit posted to IMVU's chat", len(r.products))
	x.Body = body
	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 ("/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()
	var conns []*proxy.IMQConn
	for c := range p.room.conns {
		conns = append(conns, c)
	}
	p.roomMu.Unlock()
	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)
		}
	}
}