lislgosms/gateway/internal/inbound/sessions.go

package inbound

import (
	"context"
	cmpp "github.com/bigwhite/gocmpp"
	"log"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

// This registry is the single in-process owner of authenticated downstream
// sessions. Delivery and recovery code must resolve sessions through it.

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
	tenantID       string
	applicationID  string
	enterpriseCode string
	protocol       string
	windowSize     int
	submitInFlight *atomic.Int64
	srcID          string
	phoneNumber    string
	gatewayMsgID   uint64
	remoteIP       string
	connectedAt    time.Time
	connectionID   string
	conn           *cmpp.Conn
	mu             *sync.Mutex
	presence       PresenceStore
	instanceID     string
	report         func(*downstreamSession, string, string)
	deliveryReport func(downstreamDeliveryLifecycleEvent)
	protocolLog    func(protocolLogEvent)
}

var downstreamRegistry = struct {
	sync.RWMutex
	byMessageID map[string]*downstreamSession
	byAccount   map[string]*downstreamSession
	byConn      map[*cmpp.Conn]*downstreamSession
}{
	byMessageID: make(map[string]*downstreamSession),
	byAccount:   make(map[string]*downstreamSession),
	byConn:      make(map[*cmpp.Conn]*downstreamSession),
}

func (s Server) handleActivity(_ *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) {
	session := findSessionByConn(packet.Conn)
	if session == nil {
		return true, nil
	}
	switch response := packet.Packer.(type) {
	case *cmpp.CmppActiveTestReqPkt, *cmpp.CmppActiveTestRspPkt:
		if session.report != nil {
			go s.reportConnectionOrDisconnect(session, "heartbeat", "")
		}
	case *cmpp.Cmpp2DeliverRspPkt:
		handleDownstreamAcknowledgement(packet.Conn, response.SeqId, response.MsgId, uint32(response.Result), logger)
	case *cmpp.Cmpp3DeliverRspPkt:
		handleDownstreamAcknowledgement(packet.Conn, response.SeqId, response.MsgId, response.Result, logger)
	}
	return true, nil
}

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 nil
	}
	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)
		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) {
	downstreamSubmitBarrier.Lock()
	delete(downstreamSubmitBarrier.byConn, conn)
	downstreamSubmitBarrier.Unlock()
	session := findSessionByConn(conn)
	if session == nil {
		return
	}
	forgetDownstream(session)
	_ = s.reportConnectionChecked(session, "disconnected", "CMPP client connection closed")
}

func findSessionByConn(conn *cmpp.Conn) *downstreamSession {
	downstreamRegistry.RLock()
	defer downstreamRegistry.RUnlock()
	return downstreamRegistry.byConn[conn]
}

func submitWindowByConn(conn *cmpp.Conn) int {
	session := findSessionByConn(conn)
	if session == nil || session.windowSize < 1 {
		return 1
	}
	return session.windowSize
}

func beginInboundSubmit(session *downstreamSession) func() {
	if session == nil || session.submitInFlight == nil {
		return func() {}
	}
	session.submitInFlight.Add(1)
	return func() { session.submitInFlight.Add(-1) }
}

func SubmitSlotSnapshot() (int, int64) {
	downstreamRegistry.RLock()
	defer downstreamRegistry.RUnlock()
	configured := 0
	var inFlight int64
	for _, session := range downstreamRegistry.byConn {
		if session == nil {
			continue
		}
		configured += max(1, session.windowSize)
		if session.submitInFlight != nil {
			inFlight += session.submitInFlight.Load()
		}
	}
	return configured, inFlight
}

func rememberDownstream(session downstreamSession) {
	if session.messageID == "" || session.conn == nil {
		return
	}
	session.touchPresence("connected", true, false)
	downstreamRegistry.Lock()
	if existing := downstreamRegistry.byMessageID[session.messageID]; existing != nil && existing.conn == session.conn {
		// Keep a stable message-level lookup for the live connection. Receipt
		// delivery no longer uses this value for long-message fragments; it
		// reconstructs every fragment Msg_Id from that fragment's Sequence_Id.
		session.gatewayMsgID = existing.gatewayMsgID
	}
	downstreamRegistry.byMessageID[session.messageID] = &session
	downstreamRegistry.byConn[session.conn] = &session
	if session.account != "" {
		downstreamRegistry.byAccount[session.account] = &session
	}
	downstreamRegistry.Unlock()
}

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()
	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) {
	if session == nil {
		return
	}
	downstreamRegistry.Lock()
	for messageID, current := range downstreamRegistry.byMessageID {
		if current != nil && current.conn == session.conn {
			delete(downstreamRegistry.byMessageID, messageID)
		}
	}
	if session.account != "" {
		if current := downstreamRegistry.byAccount[session.account]; current == session {
			delete(downstreamRegistry.byAccount, session.account)
		}
	}
	if current := downstreamRegistry.byConn[session.conn]; current == session {
		delete(downstreamRegistry.byConn, session.conn)
	}
	downstreamRegistry.Unlock()
	_ = session.removePresence()
}

func onlineAccounts() []string {
	downstreamRegistry.RLock()
	defer downstreamRegistry.RUnlock()
	accounts := make([]string, 0, len(downstreamRegistry.byAccount))
	for account := range downstreamRegistry.byAccount {
		if strings.TrimSpace(account) != "" {
			accounts = append(accounts, account)
		}
	}
	return accounts
}

// ActiveConnectionCount exposes only an aggregate gauge; account and remote-IP labels are intentionally excluded.
func ActiveConnectionCount() int {
	downstreamRegistry.RLock()
	defer downstreamRegistry.RUnlock()
	return len(downstreamRegistry.byConn)
}

// DisconnectAccount closes every live downstream CMPP session for an
// application account. The normal connection-close callback removes registry
// and presence state and reports the disconnect to the API.
func DisconnectAccount(account string) int {
	account = strings.TrimSpace(account)
	if account == "" {
		return 0
	}
	downstreamRegistry.RLock()
	sessions := make([]*downstreamSession, 0)
	for _, session := range downstreamRegistry.byConn {
		if session != nil && session.account == account && session.conn != nil {
			sessions = append(sessions, session)
		}
	}
	downstreamRegistry.RUnlock()
	for _, session := range sessions {
		session.conn.Close()
	}
	return len(sessions)
}

func (s Server) gatewayInstanceID() string {
	if strings.TrimSpace(s.GatewayInstanceID) != "" {
		return strings.TrimSpace(s.GatewayInstanceID)
	}
	return "gateway-1"
}

func (session downstreamSession) touchPresence(state string, includeSubmit bool, includeDeliver bool) {
	if session.presence == nil || strings.TrimSpace(session.account) == "" {
		return
	}
	now := time.Now().UTC()
	snapshot := DownstreamPresence{
		Account:           strings.TrimSpace(session.account),
		SrcID:             strings.TrimSpace(session.srcID),
		RemoteIP:          strings.TrimSpace(session.remoteIP),
		GatewayInstanceID: strings.TrimSpace(session.instanceID),
		State:             defaultString(strings.TrimSpace(state), "connected"),
		ConnectedAt:       session.connectedAt,
		UpdatedAt:         now,
	}
	if includeSubmit {
		snapshot.LastSubmitAt = now
	}
	if includeDeliver {
		snapshot.LastDeliverAt = now
	}
	_ = session.presence.TouchAccount(context.Background(), snapshot)
}

func (session downstreamSession) removePresence() error {
	if session.presence == nil || strings.TrimSpace(session.account) == "" {
		return nil
	}
	return session.presence.RemoveAccount(context.Background(), strings.TrimSpace(session.account))
}