744 lines
23 KiB
Go
744 lines
23 KiB
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))
|
|
}
|