diff --git a/cmd/t5de-proxy/main.go b/cmd/t5de-proxy/main.go index 68da798..4901dd8 100644 --- a/cmd/t5de-proxy/main.go +++ b/cmd/t5de-proxy/main.go @@ -157,6 +157,7 @@ func run(logger *log.Logger, listen, ready string, parentPID int, trace bool, pa prods.Hosted = rooms rooms.User, rooms.LocalProduct, rooms.GlobalProduct = prods.User, prods.LocalID, prods.ProductID prods.HostedOwner = rooms.Owner + rooms.UserDeliver = prods.DeliverToUser rooms.Deliver = func(c *proxy.IMQConn, m imq.Delivered) { m.Message = prods.Incoming(c, m) c.Deliver(m) diff --git a/internal/hosted/hosted.go b/internal/hosted/hosted.go index 57a497b..1a0595b 100644 --- a/internal/hosted/hosted.go +++ b/internal/hosted/hosted.go @@ -35,17 +35,24 @@ type Hub interface { // 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"` + 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"` + Privacy string `json:"privacy"` + Moderators []int64 `json:"moderators"` + Image string `json:"image"` + Favorite bool `json:"favorite"` + Score float64 `json:"score"` + Ratings int `json:"ratings"` + MyRating int `json:"myRating"` } // Participant is someone in a hosted room. @@ -65,6 +72,8 @@ type Rooms struct { // 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) + // UserDeliver gives the client a message on its own user queue (invitations). + UserDeliver func(mount string, message []byte) error // 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) @@ -75,6 +84,11 @@ type Rooms struct { 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 + + invites map[int64]*invite // by the local ID the client knows + nextInvite int64 // + watched map[int64]map[*proxy.IMQConn]string // room -> connections watching its occupancy (their queue) + occupants map[int64]map[string]bool // room -> who the watchers were last told is in it } type room struct { @@ -223,6 +237,8 @@ func (r *Rooms) FromHub(data []byte) { c.Deliver(d) } } + case "room.invite_result", "room.invite_answer", "room.invited", "room.occupancy", "room.boot_result": + r.phase2FromHub(m.Type, data) case "room.closed": r.mu.Lock() delete(r.rooms, m.Room) @@ -257,10 +273,11 @@ func (r *Rooms) Reconnected() { for _, n := range ns { r.Hub.Send(map[string]any{"type": "room.join", "room": n}) } + r.sendWatch() } // 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) { +func (r *Rooms) join(ctx context.Context, n int64, inviteID string) (joinAnswer, error) { w := make(chan joinAnswer, 1) r.mu.Lock() if r.waiting == nil { @@ -268,7 +285,11 @@ func (r *Rooms) join(ctx context.Context, n int64) (joinAnswer, error) { } r.waiting[n] = append(r.waiting[n], w) r.mu.Unlock() - if !r.Hub.Send(map[string]any{"type": "room.join", "room": n}) { + join := map[string]any{"type": "room.join", "room": n} + if inviteID != "" { + join["invite"] = inviteID + } + if !r.Hub.Send(join) { return joinAnswer{}, fmt.Errorf("the hub isn't connected") } select { @@ -317,6 +338,9 @@ func (r *Rooms) routeQueue(c *proxy.IMQConn, unsubscribe bool, queue string, opI } me := fmt.Sprintf("/user/%d", r.user()) if unsubscribe { + if kind == "occupancy" { + r.watch(c, n, queue, false) + } if kind == "chat" { r.mu.Lock() if rm, ok := r.rooms[n]; ok { @@ -344,8 +368,9 @@ func (r *Rooms) routeQueue(c *proxy.IMQConn, unsubscribe bool, queue string, opI c.Send(imq.G2cJoinedQueue, imq.JoinedQueue(me, queue, 0)) c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "node", imq.MessageMount, nil, 0)) case "occupancy": + props := r.watch(c, n, queue, true) c.Send(imq.G2cJoinedQueue, imq.JoinedQueue(me, queue, 0)) - c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "occupants", imq.StateMount, nil, 0)) + c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "occupants", imq.StateMount, props, 0)) } if withResult { c.Send(imq.G2cResult, imq.Result(opID, 0)) @@ -381,6 +406,12 @@ func (r *Rooms) Send(queue, mount string, message, original []byte, keep bool) b Message string `json:"message"` } if json.Unmarshal(message, &m) == nil { + if strings.HasPrefix(m.Message, "*boot ") { + // The line first, so the booted client gets it (it leaves on it) before the hub takes it out. + r.Hub.Send(map[string]any{"type": "room.send", "room": n, "mount": mount, "message": string(message)}) + r.boot(n, m.Message) + return true + } if seat, ok := roomSeat(m.Message, r.user()); ok { r.Hub.Send(map[string]any{"type": "room.seat", "room": n, "seat": seat}) } @@ -451,9 +482,16 @@ func (r *Rooms) Intercept(x *proxy.Exchange) error { } r.markRoomInfo(x, x.URL.Query().Get("room_id")) return nil + case "/api/favorite_rooms.php": + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if r.favoriteForm(ctx, 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"}`)) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if r.rating(ctx, x) { return nil } } @@ -520,6 +558,9 @@ func (r *Rooms) interceptXMLRPC(x *proxy.Exchange) error { args, _ := call.Params[0].(xmlrpc.Struct) ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() + if r.interceptPhase2(ctx, x, call.Method, args) { + return nil + } switch call.Method { case "test.getRoomCardInfo", "test.getRoomState", "test.setRoomState": n, ok := RoomNumber(member(args, "room_instance_id")) @@ -542,7 +583,7 @@ func (r *Rooms) interceptXMLRPC(x *proxy.Exchange) error { if !ok { return nil } - x.Response = r.joinChat(ctx, n) + x.Response = r.joinChat(ctx, n, "") case "chat.getParticipants", "chat.terminateChat": chat, err := strconv.ParseInt(member(args, "chatId"), 10, 64) if err != nil { @@ -603,15 +644,15 @@ func (r *Rooms) roomCard(ctx context.Context, n int64) *http.Response { 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: "lieutenants", Value: lieutenants(rm.Moderators)}, {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: "description", Value: plain(rm.Description)}, {Name: "imageUrl", Value: rm.Image}, {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: "isFavorite", Value: rm.Favorite}, {Name: "showFavorite", Value: true}, {Name: "showRatings", Value: true}, {Name: "enableFlagging", Value: false}, {Name: "private", Value: false}, - {Name: "autoBootWhenOwnerLeaves", Value: false}, {Name: "rating", Value: float64(0)}, + {Name: "autoBootWhenOwnerLeaves", Value: false}, {Name: "rating", Value: rm.Score}, {Name: "permission", Value: ""}, {Name: "themedStatus", Value: ""}, {Name: "themeId", Value: ""}, {Name: "eligibleForThemedRoomSubmission", Value: ""}, {Name: "occupants", Value: []xmlrpc.Value{}}, {Name: "t5de_t5", Value: true}, // the client's T5 badge (its t5de-rooms patch) @@ -715,8 +756,8 @@ func asInt(v xmlrpc.Value) (int64, bool) { return 0, false } -func (r *Rooms) joinChat(ctx context.Context, n int64) *http.Response { - a, err := r.join(ctx, n) +func (r *Rooms) joinChat(ctx context.Context, n int64, inviteID string) *http.Response { + a, err := r.join(ctx, n, inviteID) if err != nil { r.logf("hosted: joining room %d: %v", n, err) return xmlrpcFault("T5DE's hub isn't reachable right now.") @@ -726,7 +767,12 @@ func (r *Rooms) joinChat(ctx context.Context, n int64) *http.Response { if msg == "" { msg = "You can't join this T5DE room." } - return xmlrpcFault(msg) + // Refused as the client's own refusals are (it shows the explanation; booted ones as booted). + reason := int64(0) + if a.Reason == "banned" { + reason = errorPublicRoomBooted + } + return xmlrpcAnswer(xmlrpc.Struct{{Name: "response", Value: "declined"}, {Name: "reason", Value: reason}, {Name: "explanation", Value: msg}}) } me := r.user() taken := map[int]bool{} @@ -787,6 +833,7 @@ func (r *Rooms) addToRoomList(x *proxy.Exchange) { firstPage := q.Get("page") == "" || q.Get("page") == "1" || q.Get("page") == "0" addHosted := firstPage && q.Get("keywords") == "" && q.Get("search") == "" recent := q.Get("recent") != "" + favorites := q.Get("favorite") != "" && q.Get("favorite") != "0" 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 { @@ -795,7 +842,7 @@ func (r *Rooms) addToRoomList(x *proxy.Exchange) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if addHosted { - for _, rm := range r.listedRooms(ctx, recent) { + for _, rm := range r.listedRooms(ctx, recent, favorites) { appendRoom(list, rm) } } @@ -806,10 +853,16 @@ func (r *Rooms) addToRoomList(x *proxy.Exchange) { // listedRooms is the hosted rooms for the list: the person's recent ones (or, if the hub can't say, those visited // this run), or all of them, busiest first. -func (r *Rooms) listedRooms(ctx context.Context, recent bool) []Room { +func (r *Rooms) listedRooms(ctx context.Context, recent, favorites bool) []Room { var out struct { Rooms []Room `json:"rooms"` } + if favorites { + if err := r.Hub.Call(ctx, "GET", "/api/v1/hosted-rooms?favorites=1", nil, &out); err != nil { + return nil + } + return out.Rooms + } if recent { if err := r.Hub.Call(ctx, "GET", "/api/v1/hosted-rooms?recent=1", nil, &out); err == nil { return out.Rooms @@ -896,9 +949,11 @@ func appendRoom(list map[string]json.RawMessage, rm Room) { "customers_room_id": strconv.FormatInt(Base+rm.N, 10), "customers_id": strconv.FormatInt(rm.OwnerImvuID, 10), "room_instance_id": instanceID(rm.OwnerImvuID, 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, + "description": plain(rm.Description), "image_url": rm.Image, "rating": strconv.FormatFloat(rm.Score, 'f', -1, 64), + "is_favorite": boolNumber(rm.Favorite), "language": "", "is_ap": adult, "is_av": "0", "is_vip": "0", "participants": []any{}, } + flag := map[string]bool{"is_ap": true, "is_av": true, "is_vip": true} for name, raw := range list { var arr []json.RawMessage if json.Unmarshal(raw, &arr) != nil { @@ -907,6 +962,9 @@ func appendRoom(list map[string]json.RawMessage, rm Room) { v, ok := known[name] if !ok { v = emptyLike(arr) + } else if flag[name] && len(arr) == 0 { + // Nothing to copy the type from (IMVU had no rooms): a number, as "0" would count as set. + v, _ = strconv.Atoi(v.(string)) } 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 { @@ -1097,3 +1155,19 @@ func (r *Rooms) markRoomInfo(x *proxy.Exchange, instance string) { return json.Marshal(info) } } + +// lieutenants is a room's moderators as the room card lists them (strings: the client compares str(cid)). +func lieutenants(ids []int64) []xmlrpc.Value { + out := []xmlrpc.Value{} + for _, id := range ids { + out = append(out, strconv.FormatInt(id, 10)) + } + return out +} + +func boolNumber(b bool) int { + if b { + return 1 + } + return 0 +} diff --git a/internal/hosted/hosted_test.go b/internal/hosted/hosted_test.go index e872c2b..84ab6c8 100644 --- a/internal/hosted/hosted_test.go +++ b/internal/hosted/hosted_test.go @@ -11,6 +11,7 @@ import ( "strings" "sync" "testing" + "time" "git.t5de.dev/t5de/t5de-proxy/internal/imq" "git.t5de.dev/t5de/t5de-proxy/internal/proxy" @@ -56,6 +57,11 @@ func (h *fakeHub) Call(ctx context.Context, method, path string, body, out any) return json.Unmarshal([]byte(`{"participants": [{"imvu_id": 88, "seat": 1, "name": "Erin<"}]}`), out) case method == "GET" && path == "/api/v1/hosted-rooms": return json.Unmarshal([]byte(`{"rooms": [{"n": 3, "name": "Lounge", "ownerImvuId": 88, "ownerName": "Erin", "maxUsers": 10, "occupancy": 2}]}`), out) + case (method == "PUT" || method == "DELETE") && (path == "/api/v1/hosted-rooms/3/favorite" || path == "/api/v1/hosted-rooms/3/rating"): + h.mu.Lock() + h.sent = append(h.sent, map[string]any{"type": method + " " + path, "body": body}) + h.mu.Unlock() + return nil case method == "PUT" && path == "/api/v1/hosted-rooms/3/scene": h.mu.Lock() h.scene = append(h.scene, body) @@ -144,7 +150,7 @@ func TestHostedRoom(t *testing.T) { 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-88-2100000009"}}); resp.Fault == nil || resp.Fault.String != "This room is full." { + if resp, _ := xmlrpcCall(t, r, "chat.imvu.com", "chat.getOrMakeChat", xmlrpc.Struct{{Name: "activity", Value: "publicroom-88-2100000009"}}); get(resp.Params[0], "response") != "declined" || get(resp.Params[0], "explanation") != "This room is full." { t.Fatalf("full room: %+v", resp) } @@ -309,3 +315,103 @@ func TestBadgesAndRoomInfo(t *testing.T) { t.Fatalf("room info %s", out) } } + +func TestPhase2(t *testing.T) { + r, h, _ := setup() + var userMsgs []string + r.UserDeliver = func(mount string, b []byte) error { userMsgs = append(userMsgs, mount+" "+string(b)); return nil } + call := func(method string, args xmlrpc.Struct) xmlrpc.Value { + t.Helper() + resp, ok := xmlrpcCall(t, r, "chat.imvu.com", method, args) + if !ok { + t.Fatalf("%s left to IMVU", method) + } + return resp.Params[0] + } + xmlrpcCall(t, r, "chat.imvu.com", "chat.getOrMakeChat", xmlrpc.Struct{{Name: "activity", Value: "publicroom-88-2100000003"}}) + + // Inviting: waiting until the hub says it was accepted. + first := call("chat.attemptInvite", xmlrpc.Struct{{Name: "partnerId", Value: int64(55)}, {Name: "location", Value: "88-2100000003"}}) + id := get(first, "inviteId") + if get(first, "waiting") == nil || len(h.sentOf("room.invite")) != 1 || h.sentOf("room.invite")[0]["imvu_id"] != int64(55) { + t.Fatalf("first attempt %v, sent %v", first, h.sentOf("room.invite")) + } + r.FromHub([]byte(`{"type": "room.invite_result", "room": 3, "imvu_id": 55, "ok": true, "invite": "abc"}`)) + if again := call("chat.attemptInvite", xmlrpc.Struct{{Name: "inviteId", Value: id}, {Name: "location", Value: "88-2100000003"}}); get(again, "waiting") == nil { + t.Fatalf("still waiting expected: %v", again) + } + r.FromHub([]byte(`{"type": "room.invite_answer", "invite": "abc", "to": 55, "accept": true}`)) + if done := call("chat.attemptInvite", xmlrpc.Struct{{Name: "inviteId", Value: id}, {Name: "location", Value: "88-2100000003"}}); get(done, "waiting") != nil || get(done, "response") != nil { + t.Fatalf("accepted expected: %v", done) + } + + // Invited: the client gets chatInvite on its user queue; accepting joins with the invite. + r.FromHub([]byte(`{"type": "room.invited", "room": 3, "invite": "xyz", "from": 88, "room_name": "Lounge"}`)) + for i := 0; i < 100 && len(userMsgs) == 0; i++ { + time.Sleep(10 * time.Millisecond) + } + if len(userMsgs) != 1 || !strings.HasPrefix(userMsgs[0], "web_msg chatInvite ") || !strings.Contains(userMsgs[0], `"location":"88-2100000003"`) || !strings.Contains(userMsgs[0], `"name":"Erin"`) { + t.Fatalf("invitation %v", userMsgs) + } + var inv map[string]any + json.Unmarshal([]byte(strings.TrimPrefix(userMsgs[0], "web_msg chatInvite ")), &inv) + accepted := call("chat.acceptInvite", xmlrpc.Struct{{Name: "inviteId", Value: int64(inv["inviteId"].(float64))}, {Name: "chatId", Value: int64(2100000003)}}) + if get(accepted, "chatId") != int64(2100000003) || h.sentOf("room.invite_reply")[0]["accept"] != true || h.sentOf("room.join")[len(h.sentOf("room.join"))-1]["invite"] != "xyz" { + t.Fatalf("accepting: %v, sent %v", accepted, h.sent) + } + + // Booting: the *boot line also tells the hub; bootOutOfChat is answered here. + r.Send("/chat/2100000003", "messages", []byte(`{"message": "*boot 7 88", "userId": 7}`), nil, false) + if boots := h.sentOf("room.boot"); len(boots) != 1 || boots[0]["imvu_id"] != int64(88) { + t.Fatalf("boot %v", boots) + } + call("chat.bootOutOfChat", xmlrpc.Struct{{Name: "userId", Value: int64(88)}, {Name: "roomInstanceId", Value: "88-2100000003"}}) + + // Favorites. + resp, _ := xmlrpcCall(t, r, "client-dynamic.imvu.com", "test.addFavoriteRoom", xmlrpc.Struct{{Name: "room_instance_id", Value: "88-2100000003"}}) + if resp == nil || len(h.sentOf("PUT /api/v1/hosted-rooms/3/favorite")) != 1 { + t.Fatal("favorite not added") + } + + // Favorites from the room list. + fu, _ := url.Parse("https://client-dynamic.imvu.com/api/favorite_rooms.php") + fbody := []byte(`{"action": "remove", "room_instance_id": "88-2100000003"}`) + freq, _ := http.NewRequest("POST", fu.String(), bytes.NewReader(fbody)) + fx := &proxy.Exchange{URL: fu, Request: freq, Body: fbody} + r.Intercept(fx) + if fx.Response == nil || len(h.sentOf("DELETE /api/v1/hosted-rooms/3/favorite")) != 1 { + t.Fatal("favorite not removed") + } + + // Rating. + u, _ := url.Parse("https://client-dynamic.imvu.com/api/rooms/room_rating.php") + body := []byte("room_instance_id=88-2100000003&score=4") + 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 || len(h.sentOf("PUT /api/v1/hosted-rooms/3/rating")) != 1 { + t.Fatal("rating not sent") + } + + // Occupancy: watching tells the hub; changes reach the watcher as state changes. + var toClient bytes.Buffer + c := proxy.NewIMQConn(2, &toClient) + r.Route(c, false, imq.Subscription{Queues: []string{"room_occupancy/88-2100000003"}}) + if w := h.sentOf("room.watch"); len(w) == 0 || fmt.Sprint(w[len(w)-1]["rooms"]) != "[3]" { + t.Fatalf("watch %v", w) + } + toClient.Reset() + r.FromHub([]byte(`{"type": "room.occupancy", "room": 3, "participants": [{"imvu_id": 88, "seat": 1}, {"imvu_id": 7, "seat": 2}]}`)) + fs := frames(t, toClient.Bytes()) + if len(fs) != 1 || fs[0].Type != imq.G2cStateChange || !strings.Contains(string(fs[0].Data), "room_occupancy/88-2100000003") { + t.Fatalf("occupancy frames %v", fs) + } +} + +func TestAppendRoomToEmptyList(t *testing.T) { + list := map[string]json.RawMessage{"customers_id": json.RawMessage(`[]`), "is_ap": json.RawMessage(`[]`), "is_vip": json.RawMessage(`[]`), "name": json.RawMessage(`[]`)} + appendRoom(list, Room{N: 3, OwnerImvuID: 88, Name: "Lounge"}) + if string(list["is_ap"]) != "[0]" || string(list["is_vip"]) != "[0]" || string(list["name"]) != `["Lounge"]` { + t.Fatalf("list %s %s %s", list["is_ap"], list["is_vip"], list["name"]) + } +} diff --git a/internal/hosted/phase2.go b/internal/hosted/phase2.go new file mode 100644 index 0000000..cc1c9e6 --- /dev/null +++ b/internal/hosted/phase2.go @@ -0,0 +1,408 @@ +package hosted + +// Hosted rooms phase 2 (t5de-hub's docs/hosted-rooms.md, "Phase 2"): the client's own room actions, for hosted +// rooms. Boots are the room's "*boot " chat line (every client acts on it) and a room.boot to the +// hub; invitations go through the hub (the inviter's client polls chat.attemptInvite; the invitee's gets a +// "chatInvite" on its user queue); favorites, ratings and the room list's occupancy come from the hub. + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "strconv" + "strings" + "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" +) + +// errorPublicRoomBooted is the client's reason code for being refused a room it was booted from. +const errorPublicRoomBooted = 4 + +// invite is an invitation the client sent (outgoing, by the local ID the client knows) or got (incoming). +type invite struct { + room int64 + hub string // the hub's invite ID ("" until the hub answers) + to int64 // outgoing: who + from int64 // incoming: who + state string // outgoing: "waiting", "accepted", "declined" + note string // why it was declined +} + +func (r *Rooms) newInviteLocked(in *invite) int64 { + if r.invites == nil { + r.invites = map[int64]*invite{} + } + r.nextInvite++ + r.invites[r.nextInvite] = in + return r.nextInvite +} + +// --- from the hub --- + +func (r *Rooms) phase2FromHub(kind string, data []byte) { + var m struct { + Room int64 `json:"room"` + ImvuID int64 `json:"imvu_id"` + OK bool `json:"ok"` + Invite string `json:"invite"` + To int64 `json:"to"` + From int64 `json:"from"` + Accept bool `json:"accept"` + Offline bool `json:"offline"` + Note string `json:"note"` + Reason string `json:"reason"` + Message string `json:"message"` + RoomName string `json:"room_name"` + Participants []Participant `json:"participants"` + } + if json.Unmarshal(data, &m) != nil { + return + } + switch kind { + case "room.invite_result": + r.mu.Lock() + for _, in := range r.invites { + if in.room == m.Room && in.to == m.ImvuID && in.hub == "" && in.state == "waiting" { + if m.OK { + in.hub = m.Invite + } else { + in.state, in.note = "declined", m.Message + } + break + } + } + r.mu.Unlock() + case "room.invite_answer": + r.mu.Lock() + for _, in := range r.invites { + if in.hub == m.Invite && in.state == "waiting" { + switch { + case m.Offline: + in.state, in.note = "declined", "isn't online" + case m.Accept: + in.state = "accepted" + default: + in.state, in.note = "declined", m.Note + } + } + } + r.mu.Unlock() + case "room.invited": + go r.deliverInvite(m.Room, m.Invite, m.From) + case "room.occupancy": + r.occupancy(m.Room, m.Participants) + case "room.boot_result": + if !m.OK { + r.logf("hosted: booting %d from room %d: %s", m.ImvuID, m.Room, m.Message) + } + } +} + +// deliverInvite gives the client an invitation into a hosted room, as IMVU's server does ("chatInvite " on its +// user queue, mount web_msg). +func (r *Rooms) deliverInvite(n int64, hubID string, from int64) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + rm, _, _, err := r.getRoom(ctx, n) + if err != nil || r.UserDeliver == nil { + return + } + name := "" + for _, p := range r.people(ctx, n) { + if p.ImvuID == from { + name = p.Name + } + } + r.mu.Lock() + local := r.newInviteLocked(&invite{room: n, hub: hubID, from: from}) + r.mu.Unlock() + b, _ := json.Marshal(map[string]any{"chatId": Base + n, "inviteId": local, "partnerId": from, + "location": instanceID(rm.OwnerImvuID, n), + "inviter": map[string]any{"name": plain(name), "age": "", "gender": "", "imageUrl": "", "relationship": ""}}) + if err := r.UserDeliver("web_msg", append([]byte("chatInvite "), b...)); err != nil { + r.logf("hosted: giving the client an invitation: %v", err) + } +} + +type person struct { + ImvuID int64 `json:"imvu_id"` + Seat int `json:"seat"` + Name string `json:"name"` +} + +// people is who's in a room, with names. +func (r *Rooms) people(ctx context.Context, n int64) []person { + var out struct { + Participants []person `json:"participants"` + } + r.Hub.Call(ctx, "GET", fmt.Sprintf("/api/v1/hosted-rooms/%d/participants", n), nil, &out) + return out.Participants +} + +// --- the client's calls --- + +// interceptPhase2 answers the client's room actions for hosted rooms; false if the call isn't one. +func (r *Rooms) interceptPhase2(ctx context.Context, x *proxy.Exchange, method string, args xmlrpc.Struct) bool { + switch method { + case "chat.attemptInvite": + n, ok := RoomNumber(member(args, "location")) + if !ok { + if chat, err := strconv.ParseInt(member(args, "chatId"), 10, 64); err == nil { + n, ok = ChatRoom(chat) + } + } + if !ok { + return false + } + x.Response = r.attemptInvite(n, args) + return true + case "chat.leaveInviteAwayNote": + if r.knownInvite(member(args, "inviteId")) { + x.Response = xmlrpcAnswer(true) + return true + } + case "chat.acceptInvite", "chat.declineInvite": + local, _ := strconv.ParseInt(member(args, "inviteId"), 10, 64) + r.mu.Lock() + in, ok := r.invites[local] + r.mu.Unlock() + if !ok || in.from == 0 { + return false + } + accept := method == "chat.acceptInvite" + reply := map[string]any{"type": "room.invite_reply", "invite": in.hub, "accept": accept} + if note := member(args, "reason"); !accept && note != "" { + reply["note"] = note + } + r.Hub.Send(reply) + if !accept { + x.Response = xmlrpcAnswer(true) + return true + } + x.Response = r.joinChat(ctx, in.room, in.hub) + return true + case "chat.bootOutOfChat": + // The "*boot" chat line already told the hub (Send); the client only wants an answer. + if _, ok := RoomNumber(member(args, "roomInstanceId")); ok { + x.Response = xmlrpcAnswer(true) + return true + } + case "test.addFavoriteRoom", "test.removeFavoriteRoom": + n, ok := RoomNumber(member(args, "room_instance_id")) + if !ok { + return false + } + verb := "PUT" + if method == "test.removeFavoriteRoom" { + verb = "DELETE" + } + if err := r.Hub.Call(ctx, verb, fmt.Sprintf("/api/v1/hosted-rooms/%d/favorite", n), map[string]any{}, nil); err != nil { + x.Response = xmlrpcAnswer(xmlrpc.Struct{{Name: "error", Value: "T5DE's hub didn't take it: " + err.Error()}}) + return true + } + x.Response = xmlrpcAnswer(xmlrpc.Struct{}) + return true + } + return false +} + +func (r *Rooms) knownInvite(id string) bool { + local, err := strconv.ParseInt(id, 10, 64) + if err != nil { + return false + } + r.mu.Lock() + defer r.mu.Unlock() + _, ok := r.invites[local] + return ok +} + +// attemptInvite is the inviter's poll: the first call sends the invitation; later ones (with its inviteId) say +// whether it's been answered. +func (r *Rooms) attemptInvite(n int64, args xmlrpc.Struct) *http.Response { + local, _ := strconv.ParseInt(member(args, "inviteId"), 10, 64) + r.mu.Lock() + in, ok := r.invites[local] + if !ok { + to, _ := strconv.ParseInt(member(args, "partnerId"), 10, 64) + in = &invite{room: n, to: to, state: "waiting"} + local = r.newInviteLocked(in) + } + state, note := in.state, in.note + r.mu.Unlock() + if !ok { + r.Hub.Send(map[string]any{"type": "room.invite", "room": n, "imvu_id": in.to}) + } + switch state { + case "accepted": + return xmlrpcAnswer(xmlrpc.Struct{{Name: "inviteId", Value: local}, {Name: "chatId", Value: Base + n}}) + case "declined": + if note == "" { + note = "declined" + } + return xmlrpcAnswer(xmlrpc.Struct{{Name: "inviteId", Value: local}, {Name: "response", Value: "decline"}, {Name: "reason", Value: note}}) + } + return xmlrpcAnswer(xmlrpc.Struct{{Name: "inviteId", Value: local}, {Name: "waiting", Value: int64(1)}}) +} + +// boot tells the hub about a "*boot " line the client sent in a hosted room. +func (r *Rooms) boot(n int64, text string) { + f := strings.Fields(text) + if len(f) != 3 || f[0] != "*boot" { + return + } + booted, err := strconv.ParseInt(f[2], 10, 64) + if err != nil { + return + } + r.Hub.Send(map[string]any{"type": "room.boot", "room": n, "imvu_id": booted}) +} + +// favoriteForm answers the room list's favorite_rooms.php (action=add|remove, room_instance_id) for hosted rooms. +func (r *Rooms) favoriteForm(ctx context.Context, x *proxy.Exchange) bool { + form := params(x) + n, ok := RoomNumber(form.Get("room_instance_id")) + if !ok { + return false + } + verb := "PUT" + if form.Get("action") == "remove" { + verb = "DELETE" + } + if err := r.Hub.Call(ctx, verb, fmt.Sprintf("/api/v1/hosted-rooms/%d/favorite", n), map[string]any{}, nil); err != nil { + b, _ := json.Marshal(map[string]any{"error": "T5DE's hub didn't take it: " + err.Error()}) + x.Response = proxy.LocalResponse(200, "application/json", b) + return true + } + x.Response = proxy.LocalResponse(200, "application/json", []byte(`{"result":"success"}`)) + return true +} + +// params is a web API call's parameters: its query, then its body (JSON, as imvu.http.securePost sends, or a form). +func params(x *proxy.Exchange) url.Values { + out := x.URL.Query() + var body map[string]any + if json.Unmarshal(x.Body, &body) == nil { + for k, v := range body { + switch v := v.(type) { + case string: + out.Set(k, v) + case float64: + out.Set(k, strconv.FormatFloat(v, 'f', -1, 64)) + case bool: + out.Set(k, strconv.FormatBool(v)) + } + } + return out + } + if form, err := url.ParseQuery(string(x.Body)); err == nil { + for k, v := range form { + out[k] = v + } + } + return out +} + +// --- ratings (room_rating.php) --- + +func (r *Rooms) rating(ctx context.Context, x *proxy.Exchange) bool { + form := params(x) + n, ok := RoomNumber(form.Get("room_instance_id")) + if !ok { + return false + } + if score, err := strconv.Atoi(form.Get("score")); err == nil && x.Request.Method == "POST" { + if err := r.Hub.Call(ctx, "PUT", fmt.Sprintf("/api/v1/hosted-rooms/%d/rating", n), map[string]any{"score": score}, nil); err != nil { + x.Response = proxy.LocalResponse(200, "application/json", []byte(`{"result":"failure"}`)) + return true + } + x.Response = proxy.LocalResponse(200, "application/json", []byte(`{"result":"success"}`)) + return true + } + mine := 0 + if rm, _, _, err := r.getRoom(ctx, n); err == nil { + mine = rm.MyRating + } + b, _ := json.Marshal(map[string]any{"score": mine, "result": "success"}) + x.Response = proxy.LocalResponse(200, "application/json", b) + return true +} + +// --- occupancy (room_occupancy/ queues in the room list) --- + +// watch subscribes (or unsubscribes) a connection to a room's occupancy, and tells the hub which rooms to report. +func (r *Rooms) watch(c *proxy.IMQConn, n int64, queue string, on bool) []imq.Property { + r.mu.Lock() + if r.watched == nil { + r.watched = map[int64]map[*proxy.IMQConn]string{} + } + if on { + if r.watched[n] == nil { + r.watched[n] = map[*proxy.IMQConn]string{} + } + r.watched[n][c] = queue + } else if r.watched[n] != nil { + delete(r.watched[n], c) + if len(r.watched[n]) == 0 { + delete(r.watched, n) + } + } + var props []imq.Property + for key := range r.occupants[n] { + props = append(props, imq.Property{Key: key, Value: "1"}) + } + r.mu.Unlock() + r.sendWatch() + return props +} + +func (r *Rooms) sendWatch() { + r.mu.Lock() + rooms := []int64{} + for n := range r.watched { + rooms = append(rooms, n) + } + r.mu.Unlock() + r.Hub.Send(map[string]any{"type": "room.watch", "rooms": rooms}) +} + +// occupancy gives watchers a room's occupants as state changes: a key per person (the client counts them). +func (r *Rooms) occupancy(n int64, participants []Participant) { + now := map[string]bool{} + for _, p := range participants { + now[strconv.FormatInt(p.ImvuID, 10)] = true + } + r.mu.Lock() + if r.occupants == nil { + r.occupants = map[int64]map[string]bool{} + } + before := r.occupants[n] + r.occupants[n] = now + watchers := map[*proxy.IMQConn]string{} + for c, q := range r.watched[n] { + watchers[c] = q + } + r.mu.Unlock() + var props []imq.Property + for key := range now { + if !before[key] { + props = append(props, imq.Property{Key: key, Value: "1"}) + } + } + for key := range before { + if !now[key] { + props = append(props, imq.Property{Key: key}) // empty: gone + } + } + if len(props) == 0 { + return + } + for c, queue := range watchers { + c.Send(imq.G2cStateChange, imq.StateChange(0, "admin", queue, "occupants", props, 0)) + } +} diff --git a/internal/products/room.go b/internal/products/room.go index 48d1294..e7ca985 100644 --- a/internal/products/room.go +++ b/internal/products/room.go @@ -391,11 +391,18 @@ func (p *Products) Opened(c *proxy.IMQConn) { // 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/") web_msg mount. func (p *Products) RefreshInventory() { + if err := p.DeliverToUser("web_msg", []byte("refreshInventory {}")); err != nil { + p.logf("products: refreshing the inventory: %v", err) + } +} + +// DeliverToUser gives the client a message on its own user queue (/user/), as from the gateway. +func (p *Products) DeliverToUser(mount string, message []byte) error { p.mu.Lock() user := p.user p.mu.Unlock() if user == "" { - return + return nil } p.roomMu.Lock() p.room.init() @@ -404,10 +411,11 @@ func (p *Products) RefreshInventory() { conns = append(conns, c) } p.roomMu.Unlock() - q := "/user/" + user + var last error 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) + if err := c.Deliver(imq.Delivered{UserID: "0", Queue: "/user/" + user, Mount: mount, Message: message}); err != nil { + last = err } } + return last }