fix: track live downstream cmpp connections

gateway/internal/inbound/server.go

@@ -86,6 +86,17 @@ type DownstreamUplink struct {
 	ReceivedAt    string `json:"receivedAt,omitempty"`
 }
 
+type downstreamConnectionEvent struct {
+	Account       string `json:"account"`
+	ConnectionID  string `json:"connectionId"`
+	Status        string `json:"status"`
+	RemoteIP      string `json:"remoteIp,omitempty"`
+	Protocol      string `json:"protocol,omitempty"`
+	ConnectedAt   string `json:"connectedAt,omitempty"`
+	ObservedAt    string `json:"observedAt,omitempty"`
+	ErrorMessage  string `json:"errorMessage,omitempty"`
+}
+
 type downstreamSession struct {
 	messageID      string
 	account        string
@@ -96,10 +107,12 @@ type downstreamSession struct {
 	gatewayMsgID   uint64
 	remoteIP       string
 	connectedAt    time.Time
+	connectionID   string
 	conn           *cmpp.Conn
 	mu             *sync.Mutex
 	presence       PresenceStore
 	instanceID     string
+	report         func(*downstreamSession, string, string)
 }
 
 var downstreamRegistry = struct {
@@ -124,6 +137,7 @@ func (s Server) ListenAndServe() error {
 	return cmpp.ListenAndServe(addr, cmpp.V30, 30*time.Second, 3, s.LogWriter,
 		cmpp.HandlerFunc(s.handleLogin),
 		cmpp.HandlerFunc(s.handleSubmit),
+		cmpp.HandlerFunc(s.handleActivity),
 	)
 }
 
@@ -148,19 +162,23 @@ func (s Server) handleLogin(response *cmpp.Response, packet *cmpp.Packet, logger
 		return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnAuthFailed]
 	}
 	setInboundConnectResponse(response.Packer, 0, req.AuthSrc, auth.PasswordCipher, req.Version)
+	now := time.Now().UTC()
 	session := downstreamSession{
 		account:        strings.TrimSpace(defaultString(auth.Account, account)),
 		enterpriseCode: strings.TrimSpace(auth.EnterpriseCode),
 		protocol:       cmppVersionName(req.Version),
 		srcID:          strings.TrimSpace(auth.Account),
 		remoteIP:       remoteIP(packet.Conn.Conn.RemoteAddr()),
-		connectedAt:    time.Now().UTC(),
+		connectedAt:    now,
+		connectionID:   fmt.Sprintf("%s-%d", s.gatewayInstanceID(), now.UnixNano()),
 		conn:           packet.Conn,
 		mu:             &sync.Mutex{},
 		presence:       s.PresenceStore,
 		instanceID:     s.gatewayInstanceID(),
+		report:         s.reportConnection,
 	}
 	rememberAccount(session)
+	go s.reportConnection(&session, "connected", "")
 	go s.flushPending(defaultString(auth.Account, account), logger)
 	logger.Printf(
 		"cmpp inbound event=login_accepted protocol=%s requested_version=0x%02x response_version=0x%02x account=%s remote=%s",
@@ -249,11 +267,16 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
 		gatewayMsgID:   gatewayMsgID,
 		remoteIP:       remoteIP(remote),
 		connectedAt:    time.Now().UTC(),
+		connectionID:   session.connectionID,
 		conn:           packet.Conn,
 		mu:             &sync.Mutex{},
 		presence:       s.PresenceStore,
 		instanceID:     s.gatewayInstanceID(),
+		report:         session.report,
 	})
+	if current := findSessionByConn(packet.Conn); current != nil && current.report != nil {
+		go current.report(current, "submit", "")
+	}
 	go func() {
 		if _, err := s.flushPending(account, logger); err != nil {
 			logger.Printf("cmpp inbound event=post_submit_pending_flush_failed account=%s message_id=%s error=%q", account, result.MessageID, err.Error())
@@ -266,6 +289,33 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
 	return false, nil
 }
 
+func (s Server) handleActivity(_ *cmpp.Response, packet *cmpp.Packet, _ *log.Logger) (bool, error) {
+	session := findSessionByConn(packet.Conn)
+	if session == nil || session.report == nil {
+		return true, nil
+	}
+	switch packet.Packer.(type) {
+	case *cmpp.CmppActiveTestReqPkt, *cmpp.CmppActiveTestRspPkt:
+		go session.report(session, "heartbeat", "")
+	}
+	return true, nil
+}
+
+func (s Server) reportConnection(session *downstreamSession, status string, errorMessage string) {
+	if session == nil || strings.TrimSpace(session.account) == "" || strings.TrimSpace(session.connectionID) == "" {
+		return
+	}
+	event := downstreamConnectionEvent{
+		Account: session.account, ConnectionID: session.connectionID, Status: status,
+		RemoteIP: session.remoteIP, Protocol: session.protocol,
+		ConnectedAt: formatRFC3339Nano(session.connectedAt), ObservedAt: formatRFC3339Nano(time.Now()),
+		ErrorMessage: errorMessage,
+	}
+	if err := s.post(context.Background(), "/gateway/events/inbound/connection", event, nil); err != nil {
+		log.Printf("cmpp inbound connection state callback failed account=%s connection_id=%s status=%s err=%v", session.account, session.connectionID, status, err)
+	}
+}
+
 type inboundSubmitPacket struct {
 	protocol        string
 	pkTotal         uint8
@@ -804,10 +854,16 @@ func sendDownstream(session *downstreamSession, deliver cmpp.Packer) (bool, erro
 	session.mu.Lock()
 	defer session.mu.Unlock()
 	if err := session.conn.SendPkt(deliver, <-session.conn.SeqId); err != nil {
+		if session.report != nil {
+			go session.report(session, "disconnected", err.Error())
+		}
 		forgetDownstream(session)
 		return false, err
 	}
 	session.touchPresence("connected", false, true)
+	if session.report != nil {
+		go session.report(session, "deliver", "")
+	}
 	return true, nil
 }
 

gateway/internal/inbound/server_test.go

@@ -99,6 +99,7 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) {
 	password := "secret-hash"
 	var gotAuth authRequest
 	var gotSubmit submitRequest
+	connectionEvents := make(chan downstreamConnectionEvent, 8)
 	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		switch r.URL.Path {
 		case "/api/gateway/events/inbound/authenticate":
@@ -113,6 +114,13 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) {
 			_ = json.NewEncoder(w).Encode(submitResponse{Accepted: true, MessageID: "MSG-1"})
 		case "/api/gateway/events/downstream/pending":
 			_ = json.NewEncoder(w).Encode([]pendingDelivery{})
+		case "/api/gateway/events/inbound/connection":
+			var event downstreamConnectionEvent
+			if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
+				t.Fatalf("decode connection event: %v", err)
+			}
+			connectionEvents <- event
+			w.WriteHeader(http.StatusOK)
 		default:
 			t.Fatalf("unexpected api path: %s", r.URL.Path)
 		}
@@ -130,6 +138,14 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) {
 	if err := client.Connect(addr, account, password, 2*time.Second); err != nil {
 		t.Fatalf("connect inbound cmpp: %v", err)
 	}
+	select {
+	case event := <-connectionEvents:
+		if event.Status != "connected" || event.Account != account || event.ConnectionID == "" || event.RemoteIP == "" || event.Protocol != "cmpp30" {
+			t.Fatalf("unexpected connection event: %+v", event)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("expected downstream connected callback")
+	}
 
 	content, err := cmpputils.Utf8ToUcs2("测试入站")
 	if err != nil {
@@ -207,6 +223,8 @@ func TestInboundServerNegotiatesCMPP2AndUsesAuthenticatedAccountForSubmit(t *tes
 			_ = json.NewEncoder(w).Encode(submitResponse{Accepted: true, MessageID: "MSG-CMPP2"})
 		case "/api/gateway/events/downstream/pending":
 			_ = json.NewEncoder(w).Encode([]pendingDelivery{})
+		case "/api/gateway/events/inbound/connection":
+			w.WriteHeader(http.StatusOK)
 		default:
 			t.Fatalf("unexpected api path: %s", r.URL.Path)
 		}