lislgosms/gateway/internal/inbound/server.go

package inbound

import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"log"
	"net"
	"net/http"
	"strings"
	"sync"
	"time"

	cmpp "github.com/bigwhite/gocmpp"
	cmpputils "github.com/bigwhite/gocmpp/utils"
)

const defaultHTTPTimeout = 10 * time.Second

type Server struct {
	Addr                 string
	APIBaseURL           string
	HTTPClient           *http.Client
	PendingFlushInterval time.Duration
	PresenceStore        PresenceStore
	RecoveryStore        RecoveryStore
	GatewayInstanceID    string
}

type authRequest struct {
	Account    string `json:"account"`
	AuthSource string `json:"authSource"`
	Timestamp  uint32 `json:"timestamp"`
	RemoteIP   string `json:"remoteIp,omitempty"`
}

type submitRequest struct {
	Account     string `json:"account"`
	PhoneNumber string `json:"phoneNumber"`
	Content     string `json:"content"`
	SrcID       string `json:"srcId,omitempty"`
	DestID      string `json:"destId,omitempty"`
	SequenceID  uint32 `json:"sequenceId,omitempty"`
	RemoteIP    string `json:"remoteIp,omitempty"`
}

type submitResponse struct {
	Accepted  bool   `json:"accepted"`
	MessageID string `json:"messageId"`
}

type authResponse struct {
	PasswordCipher string `json:"passwordCipher"`
	ApplicationID  string `json:"applicationId"`
	TenantID       string `json:"tenantId"`
	Account        string `json:"account"`
}

type DownstreamReceipt struct {
	DeliveryID       string `json:"deliveryId,omitempty"`
	Account          string `json:"account,omitempty"`
	ApplicationID    string `json:"applicationId,omitempty"`
	MessageID        string `json:"messageId"`
	GatewayMessageID string `json:"gatewayMessageId,omitempty"`
	PhoneNumber      string `json:"phoneNumber,omitempty"`
	ReceiptStatus    string `json:"receiptStatus"`
	RawStatus        string `json:"rawStatus,omitempty"`
	ErrorCode        string `json:"errorCode,omitempty"`
	DeliveredAt      string `json:"deliveredAt,omitempty"`
}

type DownstreamUplink struct {
	DeliveryID    string `json:"deliveryId,omitempty"`
	Account       string `json:"account,omitempty"`
	ApplicationID string `json:"applicationId,omitempty"`
	MessageID     string `json:"messageId,omitempty"`
	PhoneNumber   string `json:"phoneNumber"`
	DestID        string `json:"destId"`
	Content       string `json:"content"`
	ReceivedAt    string `json:"receivedAt,omitempty"`
}

type downstreamSession struct {
	messageID    string
	account      string
	srcID        string
	phoneNumber  string
	gatewayMsgID uint64
	remoteIP     string
	connectedAt  time.Time
	conn         *cmpp.Conn
	mu           *sync.Mutex
	presence     PresenceStore
	instanceID   string
}

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

func (s Server) ListenAndServe() error {
	addr := s.Addr
	if addr == "" {
		addr = ":17890"
	}
	s.logRecoveryCandidates(log.Default())
	go s.recoverPendingCandidates(log.Default())
	go s.runPendingFlusher(log.Default())
	return cmpp.ListenAndServe(addr, cmpp.V30, 30*time.Second, 3, nil,
		cmpp.HandlerFunc(s.handleLogin),
		cmpp.HandlerFunc(s.handleSubmit),
	)
}

func (s Server) handleLogin(response *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) {
	req, ok := packet.Packer.(*cmpp.CmppConnReqPkt)
	if !ok {
		return true, nil
	}
	resp := response.Packer.(*cmpp.Cmpp3ConnRspPkt)
	resp.Version = 0x30
	account := strings.TrimRight(req.SrcAddr, "\x00")
	if account == "" {
		resp.Status = uint32(cmpp.ErrnoConnInvalidSrcAddr)
		return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnInvalidSrcAddr]
	}
	auth, err := s.authenticate(packet.Conn.Conn.RemoteAddr(), account, req.AuthSrc, req.Timestamp)
	if err != nil {
		logger.Printf("cmpp inbound auth failed account=%s remote=%s err=%v", account, packet.Conn.Conn.RemoteAddr(), err)
		resp.Status = uint32(cmpp.ErrnoConnAuthFailed)
		return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnAuthFailed]
	}
	authSource := []byte(req.AuthSrc)
	authISMG := md5.Sum(bytes.Join([][]byte{{byte(resp.Status)}, authSource, []byte(auth.PasswordCipher)}, nil))
	resp.AuthIsmg = string(authISMG[:])
	session := downstreamSession{
		account:     strings.TrimSpace(defaultString(auth.Account, account)),
		srcID:       strings.TrimSpace(auth.Account),
		remoteIP:    remoteIP(packet.Conn.Conn.RemoteAddr()),
		connectedAt: time.Now().UTC(),
		conn:        packet.Conn,
		mu:          &sync.Mutex{},
		presence:    s.PresenceStore,
		instanceID:  s.gatewayInstanceID(),
	}
	rememberAccount(session)
	go s.flushPending(defaultString(auth.Account, account), logger)
	logger.Printf("cmpp inbound account=%s login ok remote=%s", account, packet.Conn.Conn.RemoteAddr())
	return false, nil
}

func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) {
	req, ok := packet.Packer.(*cmpp.Cmpp3SubmitReqPkt)
	if !ok {
		return true, nil
	}
	resp := response.Packer.(*cmpp.Cmpp3SubmitRspPkt)
	account := strings.TrimRight(req.MsgSrc, "\x00")
	phone := ""
	if len(req.DestTerminalId) > 0 {
		phone = strings.TrimRight(req.DestTerminalId[0], "\x00")
	}
	content, err := decodeContent(req.MsgFmt, req.MsgContent)
	if err != nil {
		logger.Printf("cmpp inbound decode submit failed account=%s seq=%d err=%v", account, req.SeqId, err)
		resp.Result = 9
		return false, nil
	}
	result, err := s.submit(packet.Conn.Conn.RemoteAddr(), submitRequest{
		Account:     account,
		PhoneNumber: phone,
		Content:     content,
		SrcID:       req.SrcId,
		DestID:      phone,
		SequenceID:  req.SeqId,
		RemoteIP:    remoteIP(packet.Conn.Conn.RemoteAddr()),
	})
	if err != nil || !result.Accepted {
		logger.Printf("cmpp inbound submit rejected account=%s phone=%s seq=%d err=%v", account, phone, req.SeqId, err)
		resp.Result = 9
		return false, nil
	}
	resp.MsgId = messageIDFrom(result.MessageID, req.SeqId)
	resp.Result = 0
	rememberDownstream(downstreamSession{
		messageID:    result.MessageID,
		account:      account,
		srcID:        strings.TrimSpace(req.SrcId),
		phoneNumber:  phone,
		gatewayMsgID: resp.MsgId,
		remoteIP:     remoteIP(packet.Conn.Conn.RemoteAddr()),
		connectedAt:  time.Now().UTC(),
		conn:         packet.Conn,
		mu:           &sync.Mutex{},
		presence:     s.PresenceStore,
		instanceID:   s.gatewayInstanceID(),
	})
	return false, nil
}

func (s Server) authenticate(remote net.Addr, account string, authSource string, timestamp uint32) (authResponse, error) {
	payload := authRequest{
		Account:    account,
		AuthSource: base64.StdEncoding.EncodeToString([]byte(authSource)),
		Timestamp:  timestamp,
		RemoteIP:   remoteIP(remote),
	}
	var result authResponse
	err := s.post(context.Background(), "/gateway/events/inbound/authenticate", payload, &result)
	return result, err
}

func (s Server) submit(remote net.Addr, payload submitRequest) (submitResponse, error) {
	payload.RemoteIP = remoteIP(remote)
	var result submitResponse
	err := s.post(context.Background(), "/gateway/events/inbound/submit", payload, &result)
	return result, err
}

type pendingDeliveryRequest struct {
	Account string `json:"account"`
	Limit   int    `json:"limit,omitempty"`
}

type pendingDelivery struct {
	ID           string          `json:"id"`
	DeliveryType string          `json:"deliveryType"`
	Payload      json.RawMessage `json:"payload"`
}

type pendingFlushResult struct {
	Account        string
	Deliveries     int
	DeliveredCount int
	FailedCount    int
	WaitingCount   int
	LastError      string
}

func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushResult, error) {
	result := pendingFlushResult{Account: account}
	if account == "" {
		return result, nil
	}
	var deliveries []pendingDelivery
	if err := s.post(context.Background(), "/gateway/events/downstream/pending", pendingDeliveryRequest{Account: account, Limit: 100}, &deliveries); err != nil {
		logger.Printf("cmpp inbound pending delivery fetch failed account=%s err=%v", account, err)
		result.LastError = err.Error()
		return result, err
	}
	result.Deliveries = len(deliveries)
	for _, delivery := range deliveries {
		delivered, err := s.pushPendingDelivery(account, delivery)
		if err != nil {
			result.FailedCount++
			result.LastError = err.Error()
			_ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]string{
				"id":           delivery.ID,
				"errorMessage": err.Error(),
			}, nil)
			continue
		}
		if delivered {
			result.DeliveredCount++
			_ = s.post(context.Background(), "/gateway/events/downstream/delivered", map[string]string{"id": delivery.ID}, nil)
			continue
		}
		result.WaitingCount++
	}
	return result, nil
}

func (s Server) pushPendingDelivery(account string, delivery pendingDelivery) (bool, error) {
	switch delivery.DeliveryType {
	case "receipt":
		var event DownstreamReceipt
		if err := json.Unmarshal(delivery.Payload, &event); err != nil {
			return false, err
		}
		event.DeliveryID = delivery.ID
		event.Account = defaultString(event.Account, account)
		return PushReceipt(event)
	case "uplink":
		var event DownstreamUplink
		if err := json.Unmarshal(delivery.Payload, &event); err != nil {
			return false, err
		}
		event.DeliveryID = delivery.ID
		event.Account = defaultString(event.Account, account)
		return PushUplink(event)
	default:
		return false, fmt.Errorf("unsupported downstream delivery type %q", delivery.DeliveryType)
	}
}

func (s Server) post(ctx context.Context, path string, payload any, result any) error {
	client := s.HTTPClient
	if client == nil {
		client = &http.Client{Timeout: defaultHTTPTimeout}
	}
	body, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(apiBaseURL(s.APIBaseURL), "/")+path, bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("api returned %s", resp.Status)
	}
	if result != nil {
		return json.NewDecoder(resp.Body).Decode(result)
	}
	return nil
}

func decodeContent(format uint8, content string) (string, error) {
	switch format {
	case 8:
		return cmpputils.Ucs2ToUtf8(content)
	case 15:
		return cmpputils.GB18030ToUtf8(content)
	default:
		return content, nil
	}
}

func apiBaseURL(value string) string {
	if value == "" {
		return "http://127.0.0.1:3000/api"
	}
	return value
}

func remoteIP(addr net.Addr) string {
	if tcp, ok := addr.(*net.TCPAddr); ok {
		return tcp.IP.String()
	}
	host, _, err := net.SplitHostPort(addr.String())
	if err == nil {
		return host
	}
	return addr.String()
}

func messageIDFrom(value string, seq uint32) uint64 {
	hash := md5.Sum([]byte(value))
	result := uint64(seq)
	for _, item := range hash[:6] {
		result = (result << 8) + uint64(item)
	}
	if result == 0 {
		return uint64(time.Now().UnixNano())
	}
	return result
}

func rememberDownstream(session downstreamSession) {
	if session.messageID == "" || session.conn == nil {
		return
	}
	session.touchPresence("connected", true, false)
	downstreamRegistry.Lock()
	downstreamRegistry.byMessageID[session.messageID] = &session
	if session.account != "" {
		downstreamRegistry.byAccount[session.account] = &session
	}
	downstreamRegistry.Unlock()
}

func rememberAccount(session downstreamSession) {
	if session.account == "" || session.conn == nil {
		return
	}
	session.touchPresence("connected", false, false)
	downstreamRegistry.Lock()
	downstreamRegistry.byAccount[session.account] = &session
	downstreamRegistry.Unlock()
}

func forgetDownstream(session *downstreamSession) {
	if session == nil {
		return
	}
	downstreamRegistry.Lock()
	if session.messageID != "" {
		if current := downstreamRegistry.byMessageID[session.messageID]; current == session {
			delete(downstreamRegistry.byMessageID, session.messageID)
		}
	}
	if session.account != "" {
		if current := downstreamRegistry.byAccount[session.account]; current == session {
			delete(downstreamRegistry.byAccount, session.account)
		}
	}
	downstreamRegistry.Unlock()
	_ = session.removePresence()
}

func (s Server) runPendingFlusher(logger *log.Logger) {
	ticker := time.NewTicker(s.pendingFlushInterval())
	defer ticker.Stop()
	for range ticker.C {
		s.flushOnlineAccounts(logger)
		s.recoverPendingCandidates(logger)
	}
}

func (s Server) flushOnlineAccounts(logger *log.Logger) {
	for _, account := range onlineAccounts() {
		_, _ = s.flushPending(account, logger)
	}
}

func (s Server) recoverPendingCandidates(logger *log.Logger) {
	candidates, err := ListRecoveryCandidates(context.Background(), s.PresenceStore)
	if err != nil {
		logger.Printf("cmpp inbound recovery candidate refresh failed err=%v", err)
		return
	}
	for _, candidate := range candidates {
		account := strings.TrimSpace(candidate.Account)
		if account == "" {
			continue
		}
		activeRecovery := DownstreamRecoveryStatus{
			Account:           account,
			GatewayInstanceID: s.gatewayInstanceID(),
		}
		if s.RecoveryStore != nil {
			decision, recoveryErr := s.RecoveryStore.StartAccountRecovery(context.Background(), account, s.gatewayInstanceID())
			if recoveryErr != nil {
				logger.Printf("cmpp inbound recovery start failed account=%s err=%v", account, recoveryErr)
				continue
			}
			if !decision.Allowed {
				logger.Printf("cmpp inbound recovery skipped account=%s reason=%s", account, decision.SkipReason)
				decision.Status.Account = account
				decision.Status.GatewayInstanceID = s.gatewayInstanceID()
				decision.Status.State = defaultString(decision.Status.State, "failed")
				decision.Status.LastSkipReason = defaultString(decision.Status.LastSkipReason, decision.SkipReason)
				if decision.Status.FailureCategory == "" {
					decision.Status.FailureCategory = recoveryFailureCategory(decision.Status.State, "", decision.Status.LastSkipReason)
				}
				s.syncRecoveryStatus(logger, account, decision.Status)
				continue
			}
			activeRecovery = decision.Status
			activeRecovery.Account = account
			activeRecovery.GatewayInstanceID = s.gatewayInstanceID()
		}
		result, flushErr := s.flushPending(account, logger)
		if s.RecoveryStore != nil {
			status := DownstreamRecoveryStatus{
				Account:           account,
				GatewayInstanceID: s.gatewayInstanceID(),
				LockToken:         activeRecovery.LockToken,
				LockOwner:         defaultString(activeRecovery.LockOwner, s.gatewayInstanceID()),
				LockAcquiredAt:    activeRecovery.LockAcquiredAt,
				LockExpiresAt:     activeRecovery.LockExpiresAt,
				LastAttemptAt:     activeRecovery.LastAttemptAt,
			}
			switch {
			case flushErr != nil:
				status.State = "failed"
				status.LastError = flushErr.Error()
				status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason)
			case result.WaitingCount > 0 && result.DeliveredCount == 0 && result.FailedCount == 0:
				status.State = "waiting_connection"
				status.LastError = "downstream client is not connected"
				status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason)
			case result.FailedCount > 0 && result.DeliveredCount > 0:
				status.State = "partial"
				status.LastError = result.LastError
				status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason)
			default:
				status.State = "success"
				status.LastError = ""
				status.FailureCategory = ""
			}
			if err := s.RecoveryStore.CompleteAccountRecovery(context.Background(), status); err != nil {
				logger.Printf("cmpp inbound recovery completion failed account=%s err=%v", account, err)
				if err == ErrRecoveryLockLost {
					status.State = "failed"
					status.LastSkipReason = "lock_lost"
					status.FailureCategory = "lock_lost"
					s.syncRecoveryStatus(logger, account, status)
				}
			} else if persisted, err := s.RecoveryStore.GetAccountRecoveryStatus(context.Background(), account); err != nil {
				logger.Printf("cmpp inbound recovery status fetch failed account=%s err=%v", account, err)
			} else {
				s.syncRecoveryStatus(logger, account, persisted)
			}
		}
	}
}

func (s Server) syncRecoveryStatus(logger *log.Logger, account string, status DownstreamRecoveryStatus) {
	if err := s.post(context.Background(), "/gateway/events/downstream/recovery-status", map[string]any{
		"account":           status.Account,
		"gatewayInstanceId": status.GatewayInstanceID,
		"state":             status.State,
		"lockOwner":         status.LockOwner,
		"lockExpiresAt":     formatRFC3339Nano(status.LockExpiresAt),
		"lastAttemptAt":     formatRFC3339Nano(status.LastAttemptAt),
		"lastSuccessAt":     formatRFC3339Nano(status.LastSuccessAt),
		"lastFailureAt":     formatRFC3339Nano(status.LastFailureAt),
		"nextRetryAt":       formatRFC3339Nano(status.NextRetryAt),
		"attemptCount":      status.AttemptCount,
		"failureCategory":   status.FailureCategory,
		"lastError":         status.LastError,
		"lastSkipReason":    status.LastSkipReason,
	}, nil); err != nil {
		logger.Printf("cmpp inbound recovery status sync failed account=%s err=%v", account, err)
	}
}

func recoveryFailureCategory(state string, lastError string, lastSkipReason string) string {
	if state == "success" || state == "running" {
		return ""
	}
	if lastSkipReason == "backoff" {
		return "backoff"
	}
	if lastSkipReason == "locked" {
		return "lock_contended"
	}
	if lastSkipReason == "lock_lost" {
		return "lock_lost"
	}
	if state == "waiting_connection" {
		return "client_disconnected"
	}
	if state == "partial" {
		return "partial_delivery_failed"
	}
	if state == "failed" && strings.TrimSpace(lastError) != "" {
		return "flush_failed"
	}
	return "unknown"
}

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
}

func PushReceipt(event DownstreamReceipt) (bool, error) {
	session := findSession(event.MessageID, event.Account)
	if session == nil {
		return false, nil
	}
	stat := strings.TrimSpace(event.RawStatus)
	if stat == "" {
		stat = cmppReceiptStatus(event.ReceiptStatus)
	}
	when := time.Now()
	if event.DeliveredAt != "" {
		if parsed, err := time.Parse(time.RFC3339Nano, event.DeliveredAt); err == nil {
			when = parsed
		}
	}
	receipt := &cmpp.CmppReceiptPkt{
		MsgId:          session.gatewayMsgID,
		Stat:           stat,
		SubmitTime:     when.Format("0601021504"),
		DoneTime:       when.Format("0601021504"),
		DestTerminalId: defaultString(event.PhoneNumber, session.phoneNumber),
		SmscSequence:   uint32(time.Now().UnixNano() & 0xffffffff),
	}
	receiptBytes, err := receipt.Pack()
	if err != nil {
		return false, err
	}
	deliver := &cmpp.Cmpp3DeliverReqPkt{
		MsgId:            session.gatewayMsgID,
		DestId:           session.srcID,
		ServiceId:        "cmpp",
		MsgFmt:           0,
		SrcTerminalId:    defaultString(event.PhoneNumber, session.phoneNumber),
		RegisterDelivery: 1,
		MsgLength:        uint8(cmpp.CmppReceiptPktLen),
		MsgContent:       string(receiptBytes),
	}
	return sendDownstream(session, deliver)
}

func PushUplink(event DownstreamUplink) (bool, error) {
	session := findSession(event.MessageID, event.Account)
	if session == nil {
		return false, nil
	}
	content, err := cmpputils.Utf8ToUcs2(event.Content)
	if err != nil {
		return false, err
	}
	deliver := &cmpp.Cmpp3DeliverReqPkt{
		MsgId:            messageIDFrom(defaultString(event.MessageID, event.Account), uint32(time.Now().UnixNano())),
		DestId:           defaultString(event.DestID, session.srcID),
		ServiceId:        "cmpp",
		MsgFmt:           8,
		SrcTerminalId:    event.PhoneNumber,
		RegisterDelivery: 0,
		MsgLength:        uint8(len(content)),
		MsgContent:       content,
	}
	return sendDownstream(session, deliver)
}

func findSession(messageID string, account string) *downstreamSession {
	downstreamRegistry.RLock()
	defer downstreamRegistry.RUnlock()
	if messageID != "" {
		if session := downstreamRegistry.byMessageID[messageID]; session != nil {
			return session
		}
	}
	if account != "" {
		return downstreamRegistry.byAccount[account]
	}
	return nil
}

func sendDownstream(session *downstreamSession, deliver *cmpp.Cmpp3DeliverReqPkt) (bool, error) {
	session.mu.Lock()
	defer session.mu.Unlock()
	if err := session.conn.SendPkt(deliver, <-session.conn.SeqId); err != nil {
		forgetDownstream(session)
		return false, err
	}
	session.touchPresence("connected", false, true)
	return true, nil
}

func (s Server) pendingFlushInterval() time.Duration {
	if s.PendingFlushInterval > 0 {
		return s.PendingFlushInterval
	}
	return time.Minute
}

func (s Server) logRecoveryCandidates(logger *log.Logger) {
	candidates, err := ListRecoveryCandidates(context.Background(), s.PresenceStore)
	if err != nil {
		logger.Printf("cmpp inbound recovery candidates load failed err=%v", err)
		return
	}
	if len(candidates) == 0 {
		logger.Printf("cmpp inbound recovery candidates loaded count=0")
		return
	}
	accounts := make([]string, 0, len(candidates))
	for _, item := range candidates {
		if strings.TrimSpace(item.Account) != "" {
			accounts = append(accounts, item.Account)
		}
	}
	logger.Printf("cmpp inbound recovery candidates loaded count=%d accounts=%s", len(candidates), strings.Join(accounts, ","))
}

func cmppReceiptStatus(status string) string {
	switch strings.ToLower(strings.TrimSpace(status)) {
	case "delivered":
		return "DELIVRD"
	case "unknown":
		return "UNKNOWN"
	default:
		return "UNDELIV"
	}
}

func defaultString(value string, fallback string) string {
	if value == "" {
		return fallback
	}
	return value
}

func formatRFC3339Nano(value time.Time) string {
	if value.IsZero() {
		return ""
	}
	return value.UTC().Format(time.RFC3339Nano)
}

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))
}