feat: improve application access and money precision

gateway/internal/inbound/server.go

@@ -64,6 +64,7 @@ type authResponse struct {
 	TenantID       string `json:"tenantId"`
 	Account        string `json:"account"`
 	EnterpriseCode string `json:"enterpriseCode"`
+	MaxConnections int    `json:"maxConnections"`
 }
 
 type DownstreamReceipt struct {
@@ -179,7 +180,7 @@ func (s Server) ListenAndServe() error {
 	s.logRecoveryCandidates(log.Default())
 	go s.recoverPendingCandidates(log.Default())
 	go s.runPendingFlusher(log.Default())
-	return cmpp.ListenAndServe(addr, cmpp.V30, 30*time.Second, 3, s.LogWriter,
+	return cmpp.ListenAndServeWithClose(addr, cmpp.V30, 30*time.Second, 3, s.LogWriter, s.handleConnectionClosed,
 		cmpp.HandlerFunc(s.handleLogin),
 		cmpp.HandlerFunc(s.handleSubmit),
 		cmpp.HandlerFunc(s.handleActivity),
@@ -206,9 +207,8 @@ func (s Server) handleLogin(response *cmpp.Response, packet *cmpp.Packet, logger
 		setInboundConnectResponse(response.Packer, cmpp.ErrnoConnAuthFailed, req.AuthSrc, "", req.Version)
 		return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnAuthFailed]
 	}
-	setInboundConnectResponse(response.Packer, 0, req.AuthSrc, auth.PasswordCipher, req.Version)
 	now := time.Now().UTC()
-	session := downstreamSession{
+	session := &downstreamSession{
 		account:        strings.TrimSpace(defaultString(auth.Account, account)),
 		enterpriseCode: strings.TrimSpace(auth.EnterpriseCode),
 		protocol:       cmppVersionName(req.Version),
@@ -223,8 +223,13 @@ func (s Server) handleLogin(response *cmpp.Response, packet *cmpp.Packet, logger
 		report:         s.reportConnection,
 		deliveryReport: s.reportDownstreamDelivery,
 	}
-	rememberAccount(session)
-	go s.reportConnection(&session, "connected", "")
+	if !rememberAccount(session, auth.MaxConnections) {
+		logger.Printf("cmpp inbound auth failed account=%s remote=%s err=connection limit exceeded max=%d", account, packet.Conn.Conn.RemoteAddr(), auth.MaxConnections)
+		setInboundConnectResponse(response.Packer, cmpp.ErrnoConnOthers, req.AuthSrc, "", req.Version)
+		return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnOthers]
+	}
+	setInboundConnectResponse(response.Packer, 0, req.AuthSrc, auth.PasswordCipher, req.Version)
+	go s.reportConnectionOrDisconnect(session, "connected", "")
 	response.AfterSend = func(sendErr error) {
 		if sendErr == nil {
 			go s.flushPending(defaultString(auth.Account, account), logger)
@@ -353,7 +358,7 @@ func (s Server) handleActivity(_ *cmpp.Response, packet *cmpp.Packet, logger *lo
 	switch response := packet.Packer.(type) {
 	case *cmpp.CmppActiveTestReqPkt, *cmpp.CmppActiveTestRspPkt:
 		if session.report != nil {
-			go session.report(session, "heartbeat", "")
+			go s.reportConnectionOrDisconnect(session, "heartbeat", "")
 		}
 	case *cmpp.Cmpp2DeliverRspPkt:
 		handleDownstreamAcknowledgement(packet.Conn, response.SeqId, response.MsgId, uint32(response.Result), logger)
@@ -364,8 +369,12 @@ func (s Server) handleActivity(_ *cmpp.Response, packet *cmpp.Packet, logger *lo
 }
 
 func (s Server) reportConnection(session *downstreamSession, status string, errorMessage string) {
+	_ = s.reportConnectionChecked(session, status, errorMessage)
+}
+
+func (s Server) reportConnectionChecked(session *downstreamSession, status string, errorMessage string) error {
 	if session == nil || strings.TrimSpace(session.account) == "" || strings.TrimSpace(session.connectionID) == "" {
-		return
+		return nil
 	}
 	event := downstreamConnectionEvent{
 		Account: session.account, ConnectionID: session.connectionID, Status: status,
@@ -375,7 +384,28 @@ func (s Server) reportConnection(session *downstreamSession, status string, erro
 	}
 	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)
+		return err
 	}
+	return nil
+}
+
+func (s Server) reportConnectionOrDisconnect(session *downstreamSession, status string, errorMessage string) {
+	if err := s.reportConnectionChecked(session, status, errorMessage); err != nil && status != "disconnected" {
+		_ = s.reportConnectionChecked(session, "disconnected", err.Error())
+		forgetDownstream(session)
+		if session.conn != nil {
+			session.conn.Close()
+		}
+	}
+}
+
+func (s Server) handleConnectionClosed(conn *cmpp.Conn) {
+	session := findSessionByConn(conn)
+	if session == nil {
+		return
+	}
+	forgetDownstream(session)
+	_ = s.reportConnectionChecked(session, "disconnected", "CMPP client connection closed")
 }
 
 func (s Server) reportDownstreamDelivery(event downstreamDeliveryLifecycleEvent) {
@@ -683,15 +713,28 @@ func rememberDownstream(session downstreamSession) {
 	downstreamRegistry.Unlock()
 }
 
-func rememberAccount(session downstreamSession) {
-	if session.account == "" || session.conn == nil {
-		return
+func rememberAccount(session *downstreamSession, maxConnections int) bool {
+	if session == nil || session.account == "" || session.conn == nil {
+		return false
+	}
+	if maxConnections <= 0 {
+		maxConnections = 1
 	}
 	session.touchPresence("connected", false, false)
 	downstreamRegistry.Lock()
-	downstreamRegistry.byAccount[session.account] = &session
-	downstreamRegistry.byConn[session.conn] = &session
-	downstreamRegistry.Unlock()
+	defer downstreamRegistry.Unlock()
+	active := 0
+	for _, current := range downstreamRegistry.byConn {
+		if current != nil && current.account == session.account {
+			active++
+		}
+	}
+	if active >= maxConnections {
+		return false
+	}
+	downstreamRegistry.byAccount[session.account] = session
+	downstreamRegistry.byConn[session.conn] = session
+	return true
 }
 
 func forgetDownstream(session *downstreamSession) {