371 lines
12 KiB
Go
371 lines
12 KiB
Go
package inbound
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
cmpp "github.com/bigwhite/gocmpp"
|
|
cmpputils "github.com/bigwhite/gocmpp/utils"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// Receipt delivery requires the original Submit mapping. Falling back to an
|
|
// arbitrary account session would acknowledge a message with the wrong Msg_Id.
|
|
|
|
type DownstreamReceipt struct {
|
|
DeliveryID string `json:"deliveryId,omitempty"`
|
|
ClaimID string `json:"claimId,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"`
|
|
SubmitSequenceID *uint32 `json:"submitSequenceId,omitempty"`
|
|
SubmitGroupMessageID string `json:"submitGroupMessageId,omitempty"`
|
|
DeliveredAt string `json:"deliveredAt,omitempty"`
|
|
}
|
|
|
|
type DownstreamUplink struct {
|
|
DeliveryID string `json:"deliveryId,omitempty"`
|
|
ClaimID string `json:"claimId,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 DownstreamSendResult struct {
|
|
Sent bool `json:"sent"`
|
|
Retryable bool `json:"retryable"`
|
|
ReasonCode string `json:"reasonCode,omitempty"`
|
|
ErrorMessage string `json:"errorMessage,omitempty"`
|
|
ConnectionID string `json:"connectionId,omitempty"`
|
|
SequenceID string `json:"sequenceId,omitempty"`
|
|
MessageID string `json:"messageId,omitempty"`
|
|
SentAt string `json:"sentAt,omitempty"`
|
|
AckDeadlineAt string `json:"ackDeadlineAt,omitempty"`
|
|
}
|
|
|
|
type downstreamDeliveryLifecycleEvent struct {
|
|
Kind string
|
|
DeliveryID string
|
|
ClaimID string
|
|
ConnectionID string
|
|
SequenceID uint32
|
|
MessageID uint64
|
|
Result uint32
|
|
ObservedAt time.Time
|
|
AckDeadlineAt time.Time
|
|
FailureType string
|
|
ErrorMessage string
|
|
}
|
|
|
|
func (s Server) reportDownstreamDelivery(event downstreamDeliveryLifecycleEvent) {
|
|
if strings.TrimSpace(event.DeliveryID) == "" {
|
|
return
|
|
}
|
|
payload := map[string]any{
|
|
"id": event.DeliveryID, "claimId": event.ClaimID, "connectionId": event.ConnectionID,
|
|
"sequenceId": strconv.FormatUint(uint64(event.SequenceID), 10),
|
|
"messageId": strconv.FormatUint(event.MessageID, 10),
|
|
}
|
|
switch event.Kind {
|
|
case "sent":
|
|
payload["sentAt"] = formatRFC3339Nano(event.ObservedAt)
|
|
payload["ackDeadlineAt"] = formatRFC3339Nano(event.AckDeadlineAt)
|
|
_ = s.post(context.Background(), "/gateway/events/downstream/sent", payload, nil)
|
|
case "acknowledged":
|
|
payload["result"] = event.Result
|
|
payload["acknowledgedAt"] = formatRFC3339Nano(event.ObservedAt)
|
|
_ = s.post(context.Background(), "/gateway/events/downstream/acknowledged", payload, nil)
|
|
case "failed":
|
|
payload["failureType"] = event.FailureType
|
|
payload["errorMessage"] = event.ErrorMessage
|
|
_ = s.post(context.Background(), "/gateway/events/downstream/failed", payload, nil)
|
|
}
|
|
}
|
|
|
|
func PushReceipt(event DownstreamReceipt) (bool, error) {
|
|
result, err := PushReceiptWithResult(event)
|
|
return result.Sent, err
|
|
}
|
|
|
|
func PushReceiptWithResult(event DownstreamReceipt) (DownstreamSendResult, error) {
|
|
return pushReceiptWithResult(event, true)
|
|
}
|
|
|
|
func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (DownstreamSendResult, error) {
|
|
session := findReceiptSession(event.MessageID, event.Account)
|
|
if session == nil && allowRecovery {
|
|
session = recoverReceiptSession(event)
|
|
}
|
|
if session == nil {
|
|
if event.SubmitSequenceID == nil {
|
|
return DownstreamSendResult{
|
|
Retryable: false,
|
|
ReasonCode: "MISSING_SUBMIT_SEQUENCE_ID",
|
|
ErrorMessage: "历史回执缺少原 Submit Sequence_Id，无法重建 Msg_Id，系统已终止重投",
|
|
}, nil
|
|
}
|
|
if strings.TrimSpace(event.MessageID) == "" || strings.TrimSpace(event.Account) == "" {
|
|
return DownstreamSendResult{
|
|
Retryable: false,
|
|
ReasonCode: "INVALID_RECEIPT_PAYLOAD",
|
|
ErrorMessage: "状态回执缺少平台消息 ID 或客户账号，系统已终止重投",
|
|
}, nil
|
|
}
|
|
return DownstreamSendResult{
|
|
Retryable: true,
|
|
ReasonCode: "CLIENT_DISCONNECTED",
|
|
ErrorMessage: "下游客户端当前未连接，等待自动重试",
|
|
}, nil
|
|
}
|
|
if downstreamSubmitResponsePending(session.conn) {
|
|
return DownstreamSendResult{
|
|
Retryable: true,
|
|
ReasonCode: "SUBMIT_RESPONSE_PENDING",
|
|
ErrorMessage: "客户 SubmitResp 尚未完成写出，回执已保留并等待响应后投递",
|
|
}, 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
|
|
}
|
|
}
|
|
// Every inbound CMPP_SUBMIT fragment receives its own SUBMIT_RESP Msg_Id.
|
|
// Rebuild that exact client Msg_Id from the fragment's original Sequence_Id;
|
|
// the live session stores only connection state and must not collapse a long
|
|
// message back to the first fragment's Msg_Id.
|
|
receiptMessageID := downstreamReceiptMessageID(event, session)
|
|
receipt := &cmpp.CmppReceiptPkt{
|
|
MsgId: receiptMessageID,
|
|
Stat: stat,
|
|
SubmitTime: when.Format("0601021504"),
|
|
DoneTime: when.Format("0601021504"),
|
|
DestTerminalId: defaultString(event.PhoneNumber, session.phoneNumber),
|
|
SmscSequence: uint32(time.Now().UnixNano() & 0xffffffff),
|
|
}
|
|
version := cmpp.V30
|
|
if session.protocol == "cmpp20" || session.protocol == "cmpp21" {
|
|
version = cmpp.V20
|
|
}
|
|
receiptBytes, err := receipt.PackVersion(version)
|
|
if err != nil {
|
|
return DownstreamSendResult{}, err
|
|
}
|
|
deliver := downstreamDeliverPacket(session, receiptMessageID, session.srcID, defaultString(event.PhoneNumber, session.phoneNumber), 0, 1, string(receiptBytes))
|
|
return sendDownstream(session, deliver, event.DeliveryID, event.ClaimID)
|
|
}
|
|
|
|
func downstreamReceiptMessageID(event DownstreamReceipt, session *downstreamSession) uint64 {
|
|
if event.SubmitSequenceID != nil {
|
|
return messageIDFrom(defaultString(event.SubmitGroupMessageID, event.MessageID), *event.SubmitSequenceID)
|
|
}
|
|
if session == nil {
|
|
return 0
|
|
}
|
|
return session.gatewayMsgID
|
|
}
|
|
|
|
func findReceiptSession(messageID string, account string) *downstreamSession {
|
|
downstreamRegistry.RLock()
|
|
defer downstreamRegistry.RUnlock()
|
|
if messageID != "" {
|
|
return downstreamRegistry.byMessageID[messageID]
|
|
}
|
|
if account != "" {
|
|
return downstreamRegistry.byAccount[account]
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func recoverReceiptSession(event DownstreamReceipt) *downstreamSession {
|
|
if event.MessageID == "" || event.SubmitSequenceID == nil || event.Account == "" {
|
|
return nil
|
|
}
|
|
downstreamRegistry.RLock()
|
|
accountSession := downstreamRegistry.byAccount[event.Account]
|
|
downstreamRegistry.RUnlock()
|
|
if accountSession == nil || accountSession.conn == nil {
|
|
return nil
|
|
}
|
|
recovered := *accountSession
|
|
recovered.messageID = event.MessageID
|
|
recovered.gatewayMsgID = messageIDFrom(defaultString(event.SubmitGroupMessageID, event.MessageID), *event.SubmitSequenceID)
|
|
return &recovered
|
|
}
|
|
|
|
func PushUplink(event DownstreamUplink) (bool, error) {
|
|
result, err := PushUplinkWithResult(event)
|
|
return result.Sent, err
|
|
}
|
|
|
|
func PushUplinkWithResult(event DownstreamUplink) (DownstreamSendResult, error) {
|
|
session := findSession(event.MessageID, event.Account)
|
|
if session == nil {
|
|
return DownstreamSendResult{
|
|
Retryable: true,
|
|
ReasonCode: "CLIENT_DISCONNECTED",
|
|
ErrorMessage: "下游客户端当前未连接，等待自动重试",
|
|
}, nil
|
|
}
|
|
content, err := cmpputils.Utf8ToUcs2(event.Content)
|
|
if err != nil {
|
|
return DownstreamSendResult{}, err
|
|
}
|
|
deliver := downstreamDeliverPacket(
|
|
session,
|
|
messageIDFrom(defaultString(event.MessageID, event.Account), uint32(time.Now().UnixNano())),
|
|
defaultString(event.DestID, session.srcID),
|
|
event.PhoneNumber,
|
|
8,
|
|
0,
|
|
content,
|
|
)
|
|
return sendDownstream(session, deliver, event.DeliveryID, event.ClaimID)
|
|
}
|
|
|
|
func errorMessageWithCode(message string, code string) string {
|
|
message = strings.TrimSpace(message)
|
|
code = strings.TrimSpace(code)
|
|
if message == "" {
|
|
message = "gateway did not complete downstream delivery"
|
|
}
|
|
if code == "" {
|
|
return message
|
|
}
|
|
return fmt.Sprintf("%s (%s)", message, code)
|
|
}
|
|
|
|
func downstreamDeliverPacket(session *downstreamSession, messageID uint64, destID string, sourceTerminalID string, msgFmt uint8, registerDelivery uint8, content string) cmpp.Packer {
|
|
if session != nil && (session.protocol == "cmpp20" || session.protocol == "cmpp21") {
|
|
return &cmpp.Cmpp2DeliverReqPkt{
|
|
MsgId: messageID, DestId: destID, ServiceId: "cmpp", MsgFmt: msgFmt,
|
|
SrcTerminalId: sourceTerminalID, RegisterDelivery: registerDelivery,
|
|
MsgLength: uint8(len(content)), MsgContent: content,
|
|
}
|
|
}
|
|
return &cmpp.Cmpp3DeliverReqPkt{
|
|
MsgId: messageID, DestId: destID, ServiceId: "cmpp", MsgFmt: msgFmt,
|
|
SrcTerminalId: sourceTerminalID, RegisterDelivery: registerDelivery,
|
|
MsgLength: uint8(len(content)), MsgContent: content,
|
|
}
|
|
}
|
|
|
|
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.Packer, deliveryID string, claimID string) (DownstreamSendResult, error) {
|
|
session.mu.Lock()
|
|
defer session.mu.Unlock()
|
|
messageID := downstreamDeliverMessageID(deliver)
|
|
if messageID == 0 {
|
|
return DownstreamSendResult{}, errors.New("refusing downstream CMPP_DELIVER with Msg_Id=0")
|
|
}
|
|
sequenceID := <-session.conn.SeqId
|
|
sentAt := time.Now().UTC()
|
|
ackDeadlineAt := sentAt.Add(downstreamAckTimeout())
|
|
result := DownstreamSendResult{
|
|
ConnectionID: session.connectionID,
|
|
SequenceID: strconv.FormatUint(uint64(sequenceID), 10),
|
|
MessageID: strconv.FormatUint(messageID, 10),
|
|
SentAt: formatRFC3339Nano(sentAt),
|
|
AckDeadlineAt: formatRFC3339Nano(ackDeadlineAt),
|
|
}
|
|
tracker := registerDownstreamAck(session, deliveryID, claimID, sequenceID, messageID, ackDeadlineAt)
|
|
// A wrapped sequence cannot overwrite another unacknowledged delivery.
|
|
if deliveryID != "" && tracker == nil {
|
|
result.Retryable = true
|
|
result.ReasonCode = "SEQUENCE_IN_USE"
|
|
return result, nil
|
|
}
|
|
if err := session.conn.SendPkt(deliver, sequenceID); err != nil {
|
|
removeDownstreamAck(tracker)
|
|
session.recordDownstreamProtocol(deliver, deliveryID, sequenceID, messageID, "failed", "SEND_FAILED", err)
|
|
if session.report != nil {
|
|
go session.report(session, "disconnected", err.Error())
|
|
}
|
|
forgetDownstream(session)
|
|
result.Retryable = true
|
|
result.ReasonCode = "SEND_FAILED"
|
|
result.ErrorMessage = err.Error()
|
|
return result, nil
|
|
}
|
|
session.recordDownstreamProtocol(deliver, deliveryID, sequenceID, messageID, "success", "", nil)
|
|
result.Sent = true
|
|
if deliveryID != "" && session.deliveryReport != nil {
|
|
go session.deliveryReport(downstreamDeliveryLifecycleEvent{
|
|
Kind: "sent", DeliveryID: deliveryID, ClaimID: claimID, ConnectionID: session.connectionID,
|
|
SequenceID: sequenceID, MessageID: messageID, ObservedAt: sentAt, AckDeadlineAt: ackDeadlineAt,
|
|
})
|
|
}
|
|
session.touchPresence("connected", false, true)
|
|
if session.report != nil {
|
|
go session.report(session, "deliver", "")
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func downstreamDeliverMetadata(deliver cmpp.Packer) (string, string) {
|
|
switch packet := deliver.(type) {
|
|
case *cmpp.Cmpp2DeliverReqPkt:
|
|
if packet.RegisterDelivery == 1 {
|
|
return "deliver_receipt", packet.SrcTerminalId
|
|
}
|
|
return "deliver_uplink", packet.SrcTerminalId
|
|
case *cmpp.Cmpp3DeliverReqPkt:
|
|
if packet.RegisterDelivery == 1 {
|
|
return "deliver_receipt", packet.SrcTerminalId
|
|
}
|
|
return "deliver_uplink", packet.SrcTerminalId
|
|
default:
|
|
return "deliver", ""
|
|
}
|
|
}
|
|
|
|
func downstreamDeliverMessageID(deliver cmpp.Packer) uint64 {
|
|
switch packet := deliver.(type) {
|
|
case *cmpp.Cmpp2DeliverReqPkt:
|
|
return packet.MsgId
|
|
case *cmpp.Cmpp3DeliverReqPkt:
|
|
return packet.MsgId
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
|
|
func cmppReceiptStatus(status string) string {
|
|
switch strings.ToLower(strings.TrimSpace(status)) {
|
|
case "delivered":
|
|
return "DELIVRD"
|
|
case "unknown":
|
|
return "UNKNOWN"
|
|
default:
|
|
return "UNDELIV"
|
|
}
|
|
}
|