hectorzhao/lislgosms
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix: reconcile shared-channel receipts and protocol logs

Browse Source
This commit is contained in:
hectorzhao
2026-07-24 12:52:06 +08:00
parent 3997349c21
commit ca12f14b00
14 changed files with 1110 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/internal/inbound/protocol_log_test.go
+52
View File
@@ -0,0 +1,52 @@
package inbound
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
)
func TestSubmitResponseProtocolLoggerEmitsActualPacketDirection(t *testing.T) {
events := make(chan protocolLogEvent, 1)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/api/gateway/events/protocol-log" {
http.NotFound(w, r)
return
}
var event protocolLogEvent
if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
t.Fatalf("decode event: %v", err)
}
events <- event
_ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true})
}))
defer server.Close()
gateway := Server{APIBaseURL: server.URL + "/api", HTTPClient: server.Client()}
gateway.submitResponseProtocolLogger(
"607532",
"cmpp20",
141,
"13127620092",
"MSG-LONG-1",
736070227905294338,
0,
)(nil)
select {
case event := <-events:
if event.Protocol != "cmpp" || event.Direction != "platform_to_client" || event.EventType != "submit_resp" {
t.Fatalf("unexpected protocol event: %+v", event)
}
if event.Account != "607532" || event.MessageID != "MSG-LONG-1" || event.ResultCode != "0" {
t.Fatalf("unexpected identifiers: %+v", event)
}
if event.Detail["sequenceId"] != float64(141) && event.Detail["sequenceId"] != uint32(141) {
t.Fatalf("sequenceId = %#v", event.Detail["sequenceId"])
}
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for protocol event")
}
}
gateway/internal/inbound/server.go
+96 -4
View File
@@ -68,10 +68,27 @@ type submitResponseMessage struct {
}
type submitResponse struct {
Accepted bool `json:"accepted"`
Result uint32 `json:"result,omitempty"`
MessageID string `json:"messageId"`
Messages []submitResponseMessage `json:"messages,omitempty"`
Accepted bool `json:"accepted"`
Result uint32 `json:"result,omitempty"`
TenantID string `json:"tenantId,omitempty"`
ApplicationID string `json:"applicationId,omitempty"`
MessageID string `json:"messageId"`
Messages []submitResponseMessage `json:"messages,omitempty"`
}
type protocolLogEvent struct {
Protocol string `json:"protocol"`
Direction string `json:"direction"`
EventType string `json:"eventType"`
Status string `json:"status"`
TenantID string `json:"tenantId,omitempty"`
ApplicationID string `json:"applicationId,omitempty"`
Account string `json:"account,omitempty"`
MessageID string `json:"messageId,omitempty"`
GatewayMessageID string `json:"gatewayMessageId,omitempty"`
Phone string `json:"phone,omitempty"`
ResultCode string `json:"resultCode,omitempty"`
Detail map[string]any `json:"detail,omitempty"`
}
type authResponse struct {
@@ -271,6 +288,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
req.protocol, req.protocol, packet.Conn.Conn.RemoteAddr(), req.sequenceID, "authenticated connection session not found",
)
setInboundSubmitResponse(response.Packer, 0, 9)
response.AfterSend = s.submitResponseProtocolLogger("", req.protocol, req.sequenceID, "", "", 0, 9)
return false, nil
}
account := session.account
@@ -282,6 +300,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
fmt.Sprintf("enterprise code mismatch: expected %s", session.enterpriseCode),
)
setInboundSubmitResponse(response.Packer, 0, 9)
response.AfterSend = s.submitResponseProtocolLogger(account, defaultString(session.protocol, req.protocol), req.sequenceID, "", "", 0, 9)
return false, nil
}
phones := make([]string, len(req.destTerminalIDs))
@@ -306,6 +325,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
clientProtocol, req.protocol, account, remote, req.sequenceID, phone, err,
)
setInboundSubmitResponse(response.Packer, 0, 9)
response.AfterSend = s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, "", 0, 9)
return false, nil
}
contentHash := fmt.Sprintf("%x", md5.Sum([]byte(content)))
@@ -335,6 +355,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
clientProtocol, req.protocol, account, remote, req.sequenceID, phone, responseResult, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash, reason,
)
setInboundSubmitResponse(response.Packer, 0, responseResult)
response.AfterSend = s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, result.MessageID, 0, responseResult)
return false, nil
}
gatewayMsgID := messageIDFrom(result.MessageID, req.sequenceID)
@@ -371,6 +392,20 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
go current.report(current, "submit", "")
}
response.AfterSend = func(sendErr error) {
s.emitProtocolLog(protocolLogEvent{
Protocol: "cmpp",
Direction: "platform_to_client",
EventType: "submit_resp",
Status: protocolSendStatus(sendErr),
TenantID: result.TenantID,
ApplicationID: result.ApplicationID,
Account: account,
MessageID: result.MessageID,
GatewayMessageID: fmt.Sprint(gatewayMsgID),
Phone: phone,
ResultCode: protocolSendResultCode(sendErr, 0),
Detail: protocolSubmitResponseDetail(req.sequenceID, sendErr),
})
if sendErr != nil {
return
}
@@ -387,6 +422,63 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
return false, nil
}
func (s Server) submitResponseProtocolLogger(
account string,
protocol string,
sequenceID uint32,
phone string,
messageID string,
gatewayMessageID uint64,
result uint32,
) func(error) {
return func(sendErr error) {
s.emitProtocolLog(protocolLogEvent{
Protocol: "cmpp",
Direction: "platform_to_client",
EventType: "submit_resp",
Status: protocolSendStatus(sendErr),
Account: account,
MessageID: messageID,
GatewayMessageID: fmt.Sprint(gatewayMessageID),
Phone: phone,
ResultCode: protocolSendResultCode(sendErr, result),
Detail: protocolSubmitResponseDetail(sequenceID, sendErr),
})
}
}
func protocolSendStatus(sendErr error) string {
if sendErr != nil {
return "failed"
}
return "success"
}
func protocolSendResultCode(sendErr error, result uint32) string {
if sendErr != nil {
return "SEND_FAILED"
}
return fmt.Sprint(result)
}
func protocolSubmitResponseDetail(sequenceID uint32, sendErr error) map[string]any {
detail := map[string]any{"sequenceId": sequenceID}
if sendErr != nil {
detail["error"] = sendErr.Error()
}
return detail
}
func (s Server) emitProtocolLog(event protocolLogEvent) {
go func() {
ctx, cancel := context.WithTimeout(context.Background(), defaultHTTPTimeout)
defer cancel()
if err := s.post(ctx, "/gateway/events/protocol-log", event, nil); err != nil {
log.Printf("cmpp inbound protocol_event direction=%s event=%s status=telemetry_failed account=%s message_id=%s error=%q", event.Direction, event.EventType, event.Account, event.MessageID, err)
}
}()
}
func (s Server) handleActivity(_ *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) {
session := findSessionByConn(packet.Conn)
if session == nil {
gateway/internal/inbound/server_test.go
+10
View File
@@ -139,6 +139,8 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) {
}
acknowledgements <- event
w.WriteHeader(http.StatusOK)
case "/api/gateway/events/protocol-log":
_ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true})
default:
t.Fatalf("unexpected api path: %s", r.URL.Path)
}
@@ -316,6 +318,8 @@ func TestInboundServerForwardsLongMessageFragmentsWithoutUDHAndAcknowledgesEachS
_ = json.NewEncoder(w).Encode([]pendingDelivery{})
case "/api/gateway/events/inbound/connection":
w.WriteHeader(http.StatusOK)
case "/api/gateway/events/protocol-log":
_ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true})
default:
t.Fatalf("unexpected api path: %s", r.URL.Path)
}
@@ -410,6 +414,8 @@ func TestSubmitResponsePrecedesQueuedFailureReceipt(t *testing.T) {
}})
case "/api/gateway/events/inbound/connection", "/api/gateway/events/downstream/sent":
w.WriteHeader(http.StatusOK)
case "/api/gateway/events/protocol-log":
_ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true})
default:
t.Fatalf("unexpected api path: %s", r.URL.Path)
}
@@ -467,6 +473,8 @@ func TestDailyLimitRejectsSubmitSynchronouslyWithoutPendingReceipt(t *testing.T)
_ = json.NewEncoder(w).Encode([]pendingDelivery{})
case "/api/gateway/events/inbound/connection":
w.WriteHeader(http.StatusOK)
case "/api/gateway/events/protocol-log":
_ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true})
default:
t.Fatalf("unexpected api path: %s", r.URL.Path)
}
@@ -562,6 +570,8 @@ func TestInboundServerNegotiatesCMPP2AndUsesAuthenticatedAccountForSubmit(t *tes
_ = json.NewEncoder(w).Encode([]pendingDelivery{})
case "/api/gateway/events/inbound/connection":
w.WriteHeader(http.StatusOK)
case "/api/gateway/events/protocol-log":
_ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true})
default:
t.Fatalf("unexpected api path: %s", r.URL.Path)
}
gateway/internal/upstream/manager.go
+131 -2
View File
@@ -637,10 +637,46 @@ func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, pa
seq, err := client.SendReqPkt(pkt)
c.sendMu.Unlock()
if err != nil {
c.emitProtocolLog(protocolLogEvent{
Protocol: "cmpp",
Direction: "platform_to_channel",
EventType: "submit",
Status: "failed",
TenantID: cmd.TenantID,
ApplicationID: cmd.ApplicationID,
ChannelID: cmd.ChannelID,
Account: c.config.Account,
MessageID: cmd.MessageID,
Phone: cmd.PhoneNumber,
ResultCode: "SEND_FAILED",
PayloadBytes: len(part.MsgContent),
Detail: map[string]any{
"segmentTotal": part.PkTotal,
"segmentIndex": part.PkNumber,
},
})
c.close()
result := submitResult(cmd, 0, "", "timeout", "SEND_FAILED", err.Error())
return 0, "", result, err
}
c.emitProtocolLog(protocolLogEvent{
Protocol: "cmpp",
Direction: "platform_to_channel",
EventType: "submit",
Status: "success",
TenantID: cmd.TenantID,
ApplicationID: cmd.ApplicationID,
ChannelID: cmd.ChannelID,
Account: c.config.Account,
MessageID: cmd.MessageID,
Phone: cmd.PhoneNumber,
PayloadBytes: len(part.MsgContent),
Detail: map[string]any{
"sequenceId": seq,
"segmentTotal": part.PkTotal,
"segmentIndex": part.PkNumber,
},
})
c.mu.Lock()
c.pending[seq] = rspCh
@@ -671,6 +707,25 @@ func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, pa
errorCode = fmt.Sprint(rsp.result)
errorMessage = fmt.Sprintf("upstream submit rejected with result %d", rsp.result)
}
c.emitProtocolLog(protocolLogEvent{
Protocol: "cmpp",
Direction: "channel_to_platform",
EventType: "submit_resp",
Status: "success",
TenantID: cmd.TenantID,
ApplicationID: cmd.ApplicationID,
ChannelID: cmd.ChannelID,
Account: c.config.Account,
MessageID: cmd.MessageID,
GatewayMessageID: gatewayMessageID,
Phone: cmd.PhoneNumber,
ResultCode: fmt.Sprint(rsp.result),
Detail: map[string]any{
"sequenceId": rsp.seqID,
"segmentTotal": part.PkTotal,
"segmentIndex": part.PkNumber,
},
})
if rsp.result == 0 {
c.mu.Lock()
c.tracker[rsp.msgID] = cmd
@@ -823,10 +878,12 @@ func (c *connection) readLoop() {
ch <- submitPartResponse{seqID: p.SeqId, msgID: p.MsgId, result: p.Result}
}
case *cmpp.Cmpp2DeliverReqPkt:
_ = c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
responseErr := c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
c.emitDeliverResponse(deliverPacketFromCMPP2(p), responseErr)
c.handleDeliver(deliverPacketFromCMPP2(p))
case *cmpp.Cmpp3DeliverReqPkt:
_ = c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
responseErr := c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
c.emitDeliverResponse(deliverPacketFromCMPP3(p), responseErr)
c.handleDeliver(deliverPacketFromCMPP3(p))
case *cmpp.CmppActiveTestReqPkt:
_ = c.sendResponse(client, &cmpp.CmppActiveTestRspPkt{}, p.SeqId)
@@ -914,6 +971,78 @@ type deliverPacket struct {
msgContent string
}
type protocolLogEvent struct {
Protocol string `json:"protocol"`
Direction string `json:"direction"`
EventType string `json:"eventType"`
Status string `json:"status"`
TenantID string `json:"tenantId,omitempty"`
ApplicationID string `json:"applicationId,omitempty"`
ChannelID string `json:"channelId,omitempty"`
Account string `json:"account,omitempty"`
MessageID string `json:"messageId,omitempty"`
GatewayMessageID string `json:"gatewayMessageId,omitempty"`
Phone string `json:"phone,omitempty"`
ResultCode string `json:"resultCode,omitempty"`
PayloadBytes int `json:"payloadBytes,omitempty"`
Detail map[string]any `json:"detail,omitempty"`
}
func (c *connection) emitDeliverResponse(pkt deliverPacket, responseErr error) {
status := "success"
resultCode := "0"
detail := map[string]any{"sequenceId": pkt.seqID}
gatewayMessageID := fmt.Sprint(pkt.msgID)
messageID := ""
phone := ""
tenantID := ""
applicationID := ""
channelID := c.channelID
if pkt.registerDelivery == 1 {
var receipt cmpp.CmppReceiptPkt
if err := receipt.Unpack([]byte(pkt.msgContent)); err == nil {
gatewayMessageID = fmt.Sprint(receipt.MsgId)
phone = strings.TrimSpace(receipt.DestTerminalId)
if cmd, ok := c.commandFor(receipt.MsgId); ok {
messageID = cmd.MessageID
tenantID = cmd.TenantID
applicationID = cmd.ApplicationID
channelID = cmd.ChannelID
}
}
}
if responseErr != nil {
status = "failed"
resultCode = "SEND_FAILED"
detail["error"] = responseErr.Error()
}
c.emitProtocolLog(protocolLogEvent{
Protocol: "cmpp",
Direction: "platform_to_channel",
EventType: "deliver_resp",
Status: status,
TenantID: tenantID,
ApplicationID: applicationID,
ChannelID: channelID,
Account: c.config.Account,
MessageID: messageID,
GatewayMessageID: gatewayMessageID,
Phone: phone,
ResultCode: resultCode,
Detail: detail,
})
}
func (c *connection) emitProtocolLog(event protocolLogEvent) {
go func() {
ctx, cancel := context.WithTimeout(context.Background(), defaultHTTPTimeout)
defer cancel()
if err := postJSON(ctx, c.httpClient, c.apiBaseURL, "/gateway/events/protocol-log", event); err != nil {
log.Printf("protocol_event protocol=%s direction=%s event=%s status=telemetry_failed channel_id=%s message_id=%s error=%q", event.Protocol, event.Direction, event.EventType, event.ChannelID, event.MessageID, err)
}
}()
}
func deliverPacketFromCMPP2(pkt *cmpp.Cmpp2DeliverReqPkt) deliverPacket {
return deliverPacket{
seqID: pkt.SeqId,
gateway/internal/upstream/protocol_log_test.go
+54
View File
@@ -0,0 +1,54 @@
package upstream
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
)
func TestEmitProtocolLogPostsSafeOutboundPacketEvent(t *testing.T) {
events := make(chan protocolLogEvent, 1)
api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/gateway/events/protocol-log" {
t.Errorf("unexpected path: %s", r.URL.Path)
http.Error(w, "unexpected path", http.StatusNotFound)
return
}
var event protocolLogEvent
if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
t.Errorf("decode protocol event: %v", err)
http.Error(w, "invalid event", http.StatusBadRequest)
return
}
events <- event
w.WriteHeader(http.StatusOK)
}))
defer api.Close()
conn := &connection{apiBaseURL: api.URL, httpClient: api.Client()}
conn.emitProtocolLog(protocolLogEvent{
Protocol: "cmpp",
Direction: "platform_to_channel",
EventType: "submit",
Status: "success",
ChannelID: "channel-1",
MessageID: "MSG-1",
Phone: "18821203795",
PayloadBytes: 32,
Detail: map[string]any{"sequenceId": 7},
})
select {
case event := <-events:
if event.Direction != "platform_to_channel" || event.EventType != "submit" || event.Status != "success" {
t.Fatalf("unexpected protocol event: %+v", event)
}
if event.MessageID != "MSG-1" || event.Phone != "18821203795" || event.PayloadBytes != 32 {
t.Fatalf("unexpected event identity: %+v", event)
}
case <-time.After(time.Second):
t.Fatal("timed out waiting for protocol event")
}
}