Room channel: other T5DE users' community products on their avatars

The proxy follows the client's IMQ chat subscriptions and tells the hub which chats it's in (and the room, from chat.getOrMakeChat). For the other T5DE users the hub says are in a chat, it gives the client the community products it has packs for, as *use and *remove lines from them, and adds them to the whole outfits they send. It also takes community products out of the outfit the client posts to IMVU's chat participant endpoint, which had been sending them to IMVU. Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-29 18:14:44 -07:00
1 parent e394f1dc24
commit 93fa2f5d90
@@ -70,6 +70,12 @@ func (s *Server) serveConnect(w http.ResponseWriter, r *http.Request) {
 	toClient := &lockedWriter{w: conn}
 	down := &stream{s: s, id: id, name: "g2c"}
 	up := &stream{s: s, id: id, name: "c2g", filter: s.IMQFilter, reply: toClient}
+	if w, ok := s.IMQFilter.(IMQWatcher); ok {
+		c := &IMQConn{ID: id, w: toClient}
+		down.watcher, down.conn = w, c
+		up.watcher, up.conn = w, c
+		defer w.Closed(c)
+	}
 	// IMQ is a session: when either side closes, it's over, so both connections close.
 	var once sync.Once
 	closeBoth := func() {
@@ -101,6 +107,34 @@ type IMQFilter interface {
 	Outgoing(queue, mount string, message []byte) (out []byte, drop bool)
 }

+// IMQWatcher is an IMQFilter that also follows each IMQ connection: the queues the client subscribes to, the
+// messages the gateway delivers (which it can change), and it can deliver messages of its own.
+type IMQWatcher interface {
+	IMQFilter
+	// Subscribed and Unsubscribed see the client's subscriptions on a connection.
+	Subscribed(c *IMQConn, queues []string)
+	Unsubscribed(c *IMQConn, queues []string)
+	// Incoming returns the message to deliver instead of m's (m.Message itself to leave it).
+	Incoming(c *IMQConn, m imq.Delivered) []byte
+	// Closed: the connection is over (and with it, its subscriptions).
+	Closed(c *IMQConn)
+}
+
+// IMQConn is one IMQ connection of the client's.
+type IMQConn struct {
+	ID int64
+	w  io.Writer
+}
+
+// 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} }
+
+// 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()))
+	return err
+}
+
 // lockedWriter serializes writes, each a whole number of frames.
 type lockedWriter struct {
 	mu sync.Mutex
@@ -125,6 +159,8 @@ type stream struct {
 	pending []byte
 	filter  IMQFilter // outgoing only
 	reply   io.Writer // outgoing only: to the client
+	watcher IMQWatcher
+	conn    *IMQConn
 }

 func (st *stream) copy(dst io.Writer, src io.Reader) {
@@ -183,6 +219,30 @@ func (st *stream) pass(dst io.Writer, data []byte) error {

 // filtered is the frame to send on: as is, changed by the filter, or none (answered here).
 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 {
+				if f.Type == imq.C2gSubscribe {
+					st.watcher.Subscribed(st.conn, queues)
+				} else {
+					st.watcher.Unsubscribed(st.conn, queues)
+				}
+			}
+			return frame
+		case imq.G2cSendMessage:
+			m, err := imq.ParseDelivered(f.Data)
+			if err != nil {
+				return frame
+			}
+			out := st.watcher.Incoming(st.conn, m)
+			if bytes.Equal(out, m.Message) {
+				return frame
+			}
+			m.Message = out
+			return imq.EncodeFrame(imq.G2cSendMessage, m.Marshal())
+		}
+	}
 	if st.filter == nil || f.Type != imq.C2gSendMessage {
 		return frame
 	}
@@ -474,3 +474,102 @@ func TestInterceptor(t *testing.T) {
 		t.Fatalf("changed: %q, server got %q, %v", b, gotBody, resp.Header)
 	}
 }
+
+// testWatcher records subscriptions, uppercases delivered messages, and delivers "welcome" on each subscription.
+type testWatcher struct {
+	suffixFilter
+	mu     sync.Mutex
+	events []string
+}
+
+func (w *testWatcher) add(e string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.events = append(w.events, e)
+}
+func (w *testWatcher) Subscribed(c *IMQConn, queues []string) {
+	w.add("sub " + strings.Join(queues, ","))
+	c.Deliver(imq.Delivered{UserID: "9", Queue: queues[0], Mount: "messages", Message: []byte("welcome")})
+}
+func (w *testWatcher) Unsubscribed(c *IMQConn, queues []string) {
+	w.add("unsub " + strings.Join(queues, ","))
+}
+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 TestConnectWatchesSubscriptionsAndDeliveries(t *testing.T) {
+	gateway := httptest.NewUnstartedServer(nil)
+	gateway.StartTLS()
+	gateway.Close()
+	ln, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{Certificates: gateway.TLS.Certificates})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ln.Close()
+	delivered := imq.EncodeFrame(imq.G2cSendMessage, imq.Delivered{UserID: "/user/5", Queue: "/chat/1", Mount: "messages", Message: []byte("hello"), Sequence: 3}.Marshal())
+	go func() {
+		c, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		defer c.Close()
+		buf := make([]byte, 4096)
+		c.Read(buf) // the subscription
+		c.Write(delivered)
+		c.SetReadDeadline(time.Now().Add(2 * time.Second))
+		c.Read(buf)
+	}()
+	roots := gateway.Client().Transport.(*http.Transport).TLSClientConfig.RootCAs
+	w := &testWatcher{}
+	p, logs := newProxy(t, &Server{TLSConfig: &tls.Config{RootCAs: roots}, IMQFilter: w,
+		Dial: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			return net.Dial(network, ln.Addr().String())
+		}})
+	conn, err := net.Dial("tcp", p.Listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	fmt.Fprintf(conn, "CONNECT example.com:443 HTTP/1.1\r\nProxy-authorization: %s\r\n\r\n", auth)
+	r := bufio.NewReader(conn)
+	for line, _ := r.ReadString('\n'); line != "\r\n"; line, _ = r.ReadString('\n') {
+		if line == "" {
+			t.Fatalf("no response; log %s", logs)
+		}
+	}
+	// queues: field 1; queues_with_results: field 2 (a Subscription, name in field 1).
+	sub := []byte{0x0a, 7}
+	sub = append(sub, "/chat/1"...)
+	sub = append(sub, 0x12, 9, 0x0a, 7)
+	sub = append(sub, "/chat/2"...)
+	conn.Write(imq.EncodeFrame(imq.C2gSubscribe, sub))
+
+	want := append(
+		imq.EncodeFrame(imq.G2cSendMessage, imq.Delivered{UserID: "9", Queue: "/chat/1", Mount: "messages", Message: []byte("welcome")}.Marshal()),
+		imq.EncodeFrame(imq.G2cSendMessage, imq.Delivered{UserID: "/user/5", Queue: "/chat/1", Mount: "messages", Message: []byte("HELLO"), Sequence: 3}.Marshal())...)
+	conn.SetReadDeadline(time.Now().Add(2 * time.Second))
+	got := make([]byte, len(want))
+	if _, err := io.ReadFull(r, got); err != nil {
+		t.Fatalf("%v; log %s", err, logs)
+	}
+	if !bytes.Equal(got, want) {
+		t.Fatalf("client got %q, want %q", got, want)
+	}
+	conn.Close()
+	for i := 0; i < 200; i++ {
+		w.mu.Lock()
+		n := len(w.events)
+		w.mu.Unlock()
+		if n == 2 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if strings.Join(w.events, "; ") != "sub /chat/1,/chat/2; closed" {
+		t.Fatalf("events %q", w.events)
+	}
+}