Keep the gateway connection alive and resume hosted queues after reconnecting

The client doesn't ping while it's sending, so when the proxy answers its frames itself (everything it
sends in a hosted room), IMVU's gateway saw an idle connection and closed it after about a minute. A
ping now goes to the gateway in place of each answered frame. And after a reconnect, hosted rooms'
queues are taken out of C2gOpenFloodgates and answered here (the gateway never answered them, so the
client gave up and closed the room).

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

internal/imq/frame_test.go

@@ -97,3 +97,17 @@ func TestSubscriptionOps(t *testing.T) {
 		t.Fatalf("all %v", all)
 	}
 }
+
+func TestFloodgates(t *testing.T) {
+	state := func(q string, seq, op uint64) []byte {
+		return appendVarintField(appendVarintField(appendBytesField(nil, 1, []byte(q)), 2, seq), 3, op)
+	}
+	data := appendBytesField(appendBytesField(nil, 1, state("/user/1", 38, 19)), 1, state("/chat/2100000001", 29, 20))
+	states, err := ParseFloodgates(data)
+	if err != nil || len(states) != 2 || states[1].Queue != "/chat/2100000001" || states[1].OpID != 20 {
+		t.Fatalf("%+v %v", states, err)
+	}
+	if string(MarshalFloodgates(states)) != string(data) || string(MarshalFloodgates(states[:1])) != string(appendBytesField(nil, 1, state("/user/1", 38, 19))) {
+		t.Fatal("re-encoded differently")
+	}
+}

internal/imq/messages.go

@@ -4,7 +4,9 @@ import "errors"
 
 // Message types the proxy reads or writes.
 const (
+	C2gPing           = 1301
 	C2gSendMessage    = 1304
+	C2gOpenFloodgates = 1306
 	C2gUnsubscribe    = 1308
 	C2gSubscribe      = 1309
 	G2cResult         = 3102
@@ -314,3 +316,42 @@ func StateChange(opID uint64, userID, queue, mount string, props []Property, seq
 	b = appendBytesField(appendBytesField(appendBytesField(b, 2, []byte(userID)), 3, []byte(queue)), 4, []byte(mount))
 	return appendVarintField(appendProperties(b, 5, props), 6, sequence)
 }
+
+// ParseFloodgates reads a C2gOpenFloodgates's data (after reconnecting, the queues to resume): each queue's state
+// as a QueueOp (its op ID for a result), keeping the entry as sent.
+func ParseFloodgates(data []byte) ([]QueueOp, error) {
+	fs, err := fields(data)
+	if err != nil {
+		return nil, err
+	}
+	var out []QueueOp
+	for _, f := range fs {
+		if f.num != 1 || !f.isLen {
+			return nil, errors.New("imq: unexpected field in C2gOpenFloodgates")
+		}
+		sub, err := fields(f.bytes)
+		if err != nil {
+			return nil, err
+		}
+		q := QueueOp{raw: f.bytes}
+		for _, sf := range sub {
+			switch {
+			case sf.num == 1 && sf.isLen:
+				q.Queue = string(sf.bytes)
+			case sf.num == 3 && !sf.isLen:
+				q.OpID = sf.varint
+			}
+		}
+		out = append(out, q)
+	}
+	return out, nil
+}
+
+// MarshalFloodgates encodes a C2gOpenFloodgates's data from queue states read by ParseFloodgates.
+func MarshalFloodgates(states []QueueOp) []byte {
+	var b []byte
+	for _, q := range states {
+		b = appendBytesField(b, 1, q.raw)
+	}
+	return b
+}

internal/proxy/connect.go

@@ -253,12 +253,33 @@ func (st *stream) filtered(f imq.Frame, frame []byte) []byte {
 			}
 			rest := router.Route(st.conn, f.Type == imq.C2gUnsubscribe, sub)
 			if rest.Empty() {
-				return nil
+				return keepalive()
 			}
 			if len(rest.All()) == len(sub.All()) {
 				return frame
 			}
 			return imq.EncodeFrame(f.Type, rest.Marshal())
+		case imq.C2gOpenFloodgates:
+			// Resuming queues after a reconnect: the router's are answered here, the rest go to the gateway.
+			router, ok := st.watcher.(IMQRouter)
+			if !ok {
+				return frame
+			}
+			states, err := imq.ParseFloodgates(f.Data)
+			if err != nil {
+				return frame
+			}
+			var rest []imq.QueueOp
+			for _, q := range states {
+				if router.Route(st.conn, false, imq.Subscription{WithResults: []imq.QueueOp{q}}).Empty() {
+					continue
+				}
+				rest = append(rest, q)
+			}
+			if len(rest) == len(states) {
+				return frame
+			}
+			return imq.EncodeFrame(imq.C2gOpenFloodgates, imq.MarshalFloodgates(rest))
 		case imq.G2cSendMessage:
 			m, err := imq.ParseDelivered(f.Data)
 			if err != nil {
@@ -284,7 +305,7 @@ func (st *stream) filtered(f imq.Frame, frame []byte) []byte {
 		if st.reply != nil {
 			st.reply.Write(imq.EncodeFrame(imq.G2cResult, imq.Result(m.OpID, 0)))
 		}
-		return nil
+		return keepalive()
 	}
 	if bytes.Equal(out, m.Message) {
 		return frame
@@ -293,6 +314,11 @@ func (st *stream) filtered(f imq.Frame, frame []byte) []byte {
 	return imq.EncodeFrame(imq.C2gSendMessage, m.Marshal())
 }
 
+// keepalive is what goes to the gateway in place of a frame the proxy answered itself: a ping. The client doesn't
+// ping while it's sending, so without it the gateway sees an idle connection and closes it (in a hosted room, where
+// everything the client sends is answered here).
+func keepalive() []byte { return imq.EncodeFrame(imq.C2gPing, nil) }
+
 func (st *stream) summary() string {
 	if st.other {
 		return fmt.Sprintf("%s %dB", st.name, st.bytes)

internal/proxy/proxy_test.go

@@ -427,7 +427,8 @@ func TestConnectFiltersOutgoingMessages(t *testing.T) {
 	}
 	conn.Close()
 	got := <-received
-	want := append(send(1, "messages", "hi!"), send(3, "control", "x")...)
+	// The dropped message's place: a ping, so the gateway doesn't see an idle connection.
+	want := append(append(send(1, "messages", "hi!"), imq.EncodeFrame(imq.C2gPing, nil)...), send(3, "control", "x")...)
 	if !bytes.Equal(got, want) {
 		t.Fatalf("gateway got %q, want %q", got, want)
 	}