412 lines
13 KiB
Go
412 lines
13 KiB
Go
package upstream
|
|
|
|
import (
|
|
"cmpp-platform/gateway/internal/metrics"
|
|
"cmpp-platform/gateway/internal/queue"
|
|
"context"
|
|
"fmt"
|
|
cmpp "github.com/bigwhite/gocmpp"
|
|
cmpputils "github.com/bigwhite/gocmpp/utils"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// Segment callbacks are emitted in submission order. The aggregate result keeps
|
|
// the first sequence and Msg_Id while retaining every segment result for billing.
|
|
|
|
func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, error) {
|
|
if err := validateSubmitCommand(cmd); err != nil {
|
|
return submitResult(cmd, 0, "", "rejected", "INVALID_COMMAND", err.Error()), err
|
|
}
|
|
|
|
pool, err := m.connectionFor(cmd)
|
|
if err != nil {
|
|
return submitResult(cmd, 0, "", "rejected", "CONNECT_FAILED", err.Error()), err
|
|
}
|
|
|
|
result, err := pool.submit(ctx, cmd, func(segment queue.SubmitSegmentResult) error {
|
|
if m.SubmitSegmentPublisher == nil {
|
|
return fmt.Errorf("submit segment result publisher is required")
|
|
}
|
|
// A segment result must reach durable local storage before the next segment.
|
|
// This cannot make the supplier/Redis boundary globally atomic, but it avoids
|
|
// holding the supplier slot for an API round trip and minimizes untracked sends.
|
|
return m.SubmitSegmentPublisher.PublishSubmitSegment(ctx, cmd, segment)
|
|
})
|
|
return result, err
|
|
}
|
|
|
|
func (p *connectionPool) submit(
|
|
ctx context.Context,
|
|
cmd queue.SubmitCommand,
|
|
onSegment func(queue.SubmitSegmentResult) error,
|
|
) (final queue.SubmitResult, finalErr error) {
|
|
defer func() {
|
|
for _, segment := range final.Segments {
|
|
if segment.FirstWireSubmitAt != nil && (final.FirstWireSubmitAt == nil || segment.FirstWireSubmitAt.Before(*final.FirstWireSubmitAt)) {
|
|
final.FirstWireSubmitAt = segment.FirstWireSubmitAt
|
|
final.WireTimeSource = "gateway_write_complete"
|
|
}
|
|
}
|
|
}()
|
|
parts, err := splitSubmitContent(cmd.CMPP.MsgFmt, cmd.Content)
|
|
if err != nil {
|
|
result := submitResult(cmd, 0, "", "rejected", "ENCODE_FAILED", err.Error())
|
|
return result, err
|
|
}
|
|
|
|
var firstSequence uint32
|
|
var firstGatewayMessageID string
|
|
segments := make([]queue.SubmitSegmentResult, 0, len(parts))
|
|
for _, part := range parts {
|
|
connectionStartedAt := time.Now()
|
|
conn, release, err := p.acquireConnection(ctx)
|
|
metrics.ObserveSubmitStage("connection_wait", err == nil, time.Since(connectionStartedAt))
|
|
if err != nil {
|
|
result := submitResult(cmd, 0, "", "timeout", "WINDOW_TIMEOUT", err.Error())
|
|
result.Segments = segments
|
|
return result, err
|
|
}
|
|
supplierStartedAt := time.Now()
|
|
seq, gatewayMessageID, result, err := conn.submitPart(ctx, cmd, part)
|
|
metrics.ObserveSubmitStage("supplier_rtt", err == nil, time.Since(supplierStartedAt))
|
|
release()
|
|
segment := submitSegmentResult(part, seq, gatewayMessageID, result)
|
|
segments = append(segments, segment)
|
|
// A one-part Submit is fully represented by the aggregate event below;
|
|
// publishing an identical segment event doubles API callbacks and database
|
|
// writes without adding crash-recovery evidence. Multi-part messages still
|
|
// persist every segment before advancing to the next supplier Submit.
|
|
if onSegment != nil && len(parts) > 1 {
|
|
if publishErr := onSegment(segment); publishErr != nil {
|
|
result.Segments = segments
|
|
return result, publishErr
|
|
}
|
|
}
|
|
if firstSequence == 0 {
|
|
firstSequence = seq
|
|
}
|
|
if firstGatewayMessageID == "" {
|
|
firstGatewayMessageID = gatewayMessageID
|
|
}
|
|
if err != nil {
|
|
result.Segments = segments
|
|
return result, err
|
|
}
|
|
if result.SubmitStatus != "accepted" {
|
|
result.Segments = segments
|
|
return result, nil
|
|
}
|
|
}
|
|
|
|
result := submitResult(cmd, firstSequence, firstGatewayMessageID, "accepted", "", "")
|
|
result.Segments = segments
|
|
return result, nil
|
|
}
|
|
|
|
func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, part submitPart) (finalSequence uint32, finalID string, final queue.SubmitResult, finalErr error) {
|
|
var wireAt *time.Time
|
|
wireSource := "not_written"
|
|
defer func() { final.FirstWireSubmitAt = wireAt; final.WireTimeSource = wireSource }()
|
|
startedAt := time.Now()
|
|
defer func() {
|
|
c.mu.Lock()
|
|
c.lastSubmitRTT = time.Since(startedAt)
|
|
c.mu.Unlock()
|
|
}()
|
|
rspCh := make(chan submitPartResponse, 1)
|
|
pkt := c.submitRequestPacket(cmd, part)
|
|
|
|
c.mu.Lock()
|
|
client := c.client
|
|
closed := c.closed
|
|
c.mu.Unlock()
|
|
if closed || client == nil {
|
|
err := fmt.Errorf("supplier connection is not available")
|
|
result := submitResult(cmd, 0, "", "timeout", "CONNECTION_LOST", err.Error())
|
|
return 0, "", result, err
|
|
}
|
|
c.sendMu.Lock()
|
|
wireSource = "write_uncertain"
|
|
seq, err := client.SendReqPkt(pkt)
|
|
if err == nil {
|
|
at := time.Now().UTC()
|
|
wireAt = &at
|
|
wireSource = "gateway_write_complete"
|
|
}
|
|
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
|
|
c.mu.Unlock()
|
|
defer func() {
|
|
c.mu.Lock()
|
|
delete(c.pending, seq)
|
|
c.mu.Unlock()
|
|
}()
|
|
|
|
timeout := time.Duration(c.config.SubmitTimeoutSeconds) * time.Second
|
|
if timeout <= 0 {
|
|
timeout = defaultSubmitTimeout
|
|
}
|
|
waitCtx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
select {
|
|
case <-waitCtx.Done():
|
|
c.markSubmitFailure()
|
|
result := submitResult(cmd, seq, "", "timeout", "SUBMIT_TIMEOUT", waitCtx.Err().Error())
|
|
return seq, "", result, waitCtx.Err()
|
|
case rsp := <-rspCh:
|
|
if rsp.err != nil {
|
|
c.markSubmitFailure()
|
|
result := submitResult(cmd, seq, "", "timeout", "CONNECTION_LOST", rsp.err.Error())
|
|
return seq, "", result, rsp.err
|
|
}
|
|
gatewayMessageID := fmt.Sprint(rsp.msgID)
|
|
status := "accepted"
|
|
errorCode := ""
|
|
errorMessage := ""
|
|
if rsp.result != 0 {
|
|
status = "rejected"
|
|
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.markSubmitSuccess()
|
|
c.mu.Lock()
|
|
c.tracker[rsp.msgID] = cmd
|
|
c.mu.Unlock()
|
|
}
|
|
return seq, gatewayMessageID, submitResult(cmd, seq, gatewayMessageID, status, errorCode, errorMessage), nil
|
|
}
|
|
}
|
|
|
|
func (c *connection) submitRequestPacket(cmd queue.SubmitCommand, part submitPart) cmpp.Packer {
|
|
base := submitRequestFields{
|
|
PkTotal: part.PkTotal,
|
|
PkNumber: part.PkNumber,
|
|
TpUdhi: part.TpUdhi,
|
|
RegisteredDelivery: uint8(cmd.CMPP.RegisteredDelivery),
|
|
MsgLevel: 1,
|
|
ServiceId: cmd.CMPP.ServiceID,
|
|
FeeUserType: uint8(defaultInt(cmd.CMPP.FeeUserType, 2)),
|
|
FeeTerminalId: cmd.PhoneNumber,
|
|
MsgFmt: uint8(cmd.CMPP.MsgFmt),
|
|
MsgSrc: c.config.Account,
|
|
FeeType: defaultString(cmd.CMPP.FeeType, "02"),
|
|
FeeCode: defaultString(cmd.CMPP.FeeCode, "0"),
|
|
SrcId: cmd.CMPP.SrcID,
|
|
DestUsrTl: 1,
|
|
DestTerminalId: []string{cmd.PhoneNumber},
|
|
MsgLength: uint8(len(part.MsgContent)),
|
|
MsgContent: part.MsgContent,
|
|
}
|
|
if protocolVersion(c.config.CMPPVersion) == cmpp.V20 {
|
|
return &cmpp.Cmpp2SubmitReqPkt{
|
|
PkTotal: base.PkTotal,
|
|
PkNumber: base.PkNumber,
|
|
RegisteredDelivery: base.RegisteredDelivery,
|
|
MsgLevel: base.MsgLevel,
|
|
ServiceId: base.ServiceId,
|
|
FeeUserType: base.FeeUserType,
|
|
FeeTerminalId: base.FeeTerminalId,
|
|
TpUdhi: base.TpUdhi,
|
|
MsgFmt: base.MsgFmt,
|
|
MsgSrc: base.MsgSrc,
|
|
FeeType: base.FeeType,
|
|
FeeCode: base.FeeCode,
|
|
SrcId: base.SrcId,
|
|
DestUsrTl: base.DestUsrTl,
|
|
DestTerminalId: base.DestTerminalId,
|
|
MsgLength: base.MsgLength,
|
|
MsgContent: base.MsgContent,
|
|
}
|
|
}
|
|
return &cmpp.Cmpp3SubmitReqPkt{
|
|
PkTotal: base.PkTotal,
|
|
PkNumber: base.PkNumber,
|
|
RegisteredDelivery: base.RegisteredDelivery,
|
|
MsgLevel: base.MsgLevel,
|
|
ServiceId: base.ServiceId,
|
|
FeeUserType: base.FeeUserType,
|
|
FeeTerminalId: base.FeeTerminalId,
|
|
TpUdhi: base.TpUdhi,
|
|
MsgFmt: base.MsgFmt,
|
|
MsgSrc: base.MsgSrc,
|
|
FeeType: base.FeeType,
|
|
FeeCode: base.FeeCode,
|
|
SrcId: base.SrcId,
|
|
DestUsrTl: base.DestUsrTl,
|
|
DestTerminalId: base.DestTerminalId,
|
|
MsgLength: base.MsgLength,
|
|
MsgContent: base.MsgContent,
|
|
}
|
|
}
|
|
|
|
type submitRequestFields struct {
|
|
PkTotal uint8
|
|
PkNumber uint8
|
|
RegisteredDelivery uint8
|
|
MsgLevel uint8
|
|
ServiceId string
|
|
FeeUserType uint8
|
|
FeeTerminalId string
|
|
TpUdhi uint8
|
|
MsgFmt uint8
|
|
MsgSrc string
|
|
FeeType string
|
|
FeeCode string
|
|
SrcId string
|
|
DestUsrTl uint8
|
|
DestTerminalId []string
|
|
MsgLength uint8
|
|
MsgContent string
|
|
}
|
|
|
|
func submitResult(cmd queue.SubmitCommand, sequenceID uint32, gatewayMessageID string, status string, code string, message string) queue.SubmitResult {
|
|
if gatewayMessageID == "" {
|
|
gatewayMessageID = fmt.Sprintf("GW-%s-%d", cmd.SubmitID, time.Now().UnixNano())
|
|
}
|
|
return queue.SubmitResult{
|
|
Envelope: queue.Envelope{
|
|
SchemaVersion: queue.SchemaVersion,
|
|
MessageType: queue.MessageTypeSubmitResult,
|
|
TraceID: cmd.TraceID,
|
|
MessageID: cmd.MessageID,
|
|
ChannelID: cmd.ChannelID,
|
|
CreatedAt: time.Now().UTC(),
|
|
},
|
|
SubmitID: cmd.SubmitID,
|
|
SequenceID: sequenceID,
|
|
GatewayMessageID: gatewayMessageID,
|
|
SubmitStatus: status,
|
|
ErrorCode: code,
|
|
ErrorMessage: message,
|
|
SubmittedAt: time.Now().UTC(),
|
|
WireTimeSource: "not_written",
|
|
ReceiptRequested: cmd.CMPP.RegisteredDelivery != 0,
|
|
}
|
|
}
|
|
|
|
func submitSegmentResult(part submitPart, sequenceID uint32, gatewayMessageID string, result queue.SubmitResult) queue.SubmitSegmentResult {
|
|
if gatewayMessageID == "" {
|
|
gatewayMessageID = result.GatewayMessageID
|
|
}
|
|
return queue.SubmitSegmentResult{
|
|
FirstWireSubmitAt: result.FirstWireSubmitAt,
|
|
WireTimeSource: result.WireTimeSource,
|
|
ReceiptRequested: result.ReceiptRequested,
|
|
SegmentTotal: int(part.PkTotal),
|
|
SegmentIndex: int(part.PkNumber),
|
|
SequenceID: sequenceID,
|
|
GatewayMessageID: gatewayMessageID,
|
|
SubmitStatus: result.SubmitStatus,
|
|
ErrorCode: result.ErrorCode,
|
|
ErrorMessage: result.ErrorMessage,
|
|
SubmittedAt: result.SubmittedAt,
|
|
}
|
|
}
|
|
|
|
func validateSubmitCommand(cmd queue.SubmitCommand) error {
|
|
if cmd.MessageType != queue.MessageTypeSubmitCommand {
|
|
return fmt.Errorf("unsupported messageType %q", cmd.MessageType)
|
|
}
|
|
if cmd.MessageID == "" || cmd.ChannelID == "" || cmd.SubmitID == "" {
|
|
return fmt.Errorf("messageId, channelId and submitId are required")
|
|
}
|
|
if cmd.Upstream.GatewayHost == "" || cmd.Upstream.GatewayPort <= 0 {
|
|
return fmt.Errorf("upstream gatewayHost and gatewayPort are required")
|
|
}
|
|
if cmd.Upstream.Account == "" || cmd.Upstream.PasswordCipher == "" {
|
|
return fmt.Errorf("upstream account and passwordCipher are required")
|
|
}
|
|
if cmd.Upstream.DesiredConnections != 0 && (cmd.Upstream.DesiredConnections < 1 || cmd.Upstream.DesiredConnections > maximumConnections) {
|
|
return fmt.Errorf("upstream desiredConnections must be between 1 and %d", maximumConnections)
|
|
}
|
|
if cmd.Upstream.WindowSize != 0 && (cmd.Upstream.WindowSize < 1 || cmd.Upstream.WindowSize > maximumWindowSize) {
|
|
return fmt.Errorf("upstream windowSize must be between 1 and %d", maximumWindowSize)
|
|
}
|
|
if len(cmd.PhoneNumber) == 0 || len(cmd.Content) == 0 {
|
|
return fmt.Errorf("phoneNumber and content are required")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func encodeContent(format int, content string) (string, error) {
|
|
switch format {
|
|
case 8:
|
|
return cmpputils.Utf8ToUcs2(content)
|
|
case 15:
|
|
return cmpputils.Utf8ToGB18030(content)
|
|
default:
|
|
return content, nil
|
|
}
|
|
}
|
|
|
|
func protocolVersion(version string) cmpp.Type {
|
|
if strings.HasPrefix(version, "2") {
|
|
return cmpp.V20
|
|
}
|
|
return cmpp.V30
|
|
}
|