package upstream import ( "cmpp-platform/gateway/internal/queue" "context" "fmt" cmpp "github.com/bigwhite/gocmpp" cmpputils "github.com/bigwhite/gocmpp/utils" "log" "strings" "time" ) // Receipt packets are correlated through the per-connection Submit tracker; // mobile-originated content uses a separate long-message assembly path. type deliverPacket struct { version cmpp.Type seqID uint32 msgID uint64 destID string tpUdhi uint8 msgFmt uint8 srcTerminalID string registerDelivery uint8 msgContent string } func deliverPacketFromCMPP2(pkt *cmpp.Cmpp2DeliverReqPkt) deliverPacket { return deliverPacket{version: cmpp.V20, seqID: pkt.SeqId, msgID: pkt.MsgId, destID: pkt.DestId, tpUdhi: pkt.TpUdhi, msgFmt: pkt.MsgFmt, srcTerminalID: pkt.SrcTerminalId, registerDelivery: pkt.RegisterDelivery, msgContent: pkt.MsgContent, } } func deliverPacketFromCMPP3(pkt *cmpp.Cmpp3DeliverReqPkt) deliverPacket { return deliverPacket{version: cmpp.V30, seqID: pkt.SeqId, msgID: pkt.MsgId, destID: pkt.DestId, tpUdhi: pkt.TpUdhi, msgFmt: pkt.MsgFmt, srcTerminalID: pkt.SrcTerminalId, registerDelivery: pkt.RegisterDelivery, msgContent: pkt.MsgContent, } } func (c *connection) handleDeliver(pkt deliverPacket) error { if pkt.registerDelivery == 1 { var receipt cmpp.CmppReceiptPkt if err := receipt.UnpackVersion([]byte(pkt.msgContent), pkt.version); err != nil { log.Printf("protocol_event protocol=cmpp direction=channel_to_platform event=deliver_receipt status=parse_failed channel_id=%s sequence_id=%d packet_msg_id=%d error=%q", c.channelID, pkt.seqID, pkt.msgID, err) return err } log.Printf("protocol_event protocol=cmpp direction=channel_to_platform event=deliver_receipt status=received channel_id=%s sequence_id=%d gateway_message_id=%d raw_status=%s", c.channelID, pkt.seqID, receipt.MsgId, strings.TrimSpace(receipt.Stat)) cmd, ok := c.commandFor(receipt.MsgId) if !ok { cmd, ok = c.commandFor(pkt.msgID) } traceID := fmt.Sprintf("receipt-%d", receipt.MsgId) messageID := fmt.Sprintf("receipt-%d", receipt.MsgId) channelID := c.channelID if ok { traceID = cmd.TraceID messageID = cmd.MessageID channelID = cmd.ChannelID } event := queue.ReceiptEvent{ Envelope: queue.Envelope{ SchemaVersion: queue.SchemaVersion, MessageType: queue.MessageTypeReceiptEvent, TraceID: traceID, MessageID: messageID, ChannelID: channelID, CreatedAt: time.Now().UTC(), }, SequenceID: pkt.seqID, GatewayMessageID: fmt.Sprint(receipt.MsgId), PhoneNumber: strings.TrimSpace(receipt.DestTerminalId), ReceiptStatus: receiptStatus(receipt.Stat), RawStatus: strings.TrimSpace(receipt.Stat), DeliveredAt: time.Now().UTC(), ConnectionID: c.identity(), } if c.protocolLogPublisher != nil { c.emitProtocolLog(protocolLogEvent{Protocol: "cmpp", Direction: "channel_to_platform", EventType: "deliver_receipt", Status: "success", ChannelID: channelID, Account: c.config.Account, MessageID: messageID, GatewayMessageID: fmt.Sprint(receipt.MsgId), Phone: strings.TrimSpace(receipt.DestTerminalId), ResultCode: strings.TrimSpace(receipt.Stat), Detail: map[string]any{"sequenceId": pkt.seqID}}) } var publishErr error if c.eventPublisher != nil { publishErr = c.eventPublisher.PublishReceipt(context.Background(), event) } else { publishErr = postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/receipt/intake", event) } if err := publishErr; err != nil { log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_receipt status=forward_failed channel_id=%s sequence_id=%d gateway_message_id=%d error=%q", c.channelID, pkt.seqID, receipt.MsgId, err) return err } else { log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_receipt status=forwarded channel_id=%s sequence_id=%d gateway_message_id=%d", c.channelID, pkt.seqID, receipt.MsgId) } return nil } content, complete, err := c.decodeUplinkContent(pkt) if err != nil { log.Printf("protocol_event protocol=cmpp direction=channel_to_platform event=deliver_uplink status=decode_failed channel_id=%s sequence_id=%d packet_msg_id=%d error=%q", c.channelID, pkt.seqID, pkt.msgID, err) return err } if !complete { return nil } cmd, _ := c.commandFor(pkt.msgID) event := queue.UplinkEvent{ Envelope: queue.Envelope{ SchemaVersion: queue.SchemaVersion, MessageType: queue.MessageTypeUplinkEvent, TraceID: cmd.TraceID, MessageID: cmd.MessageID, ChannelID: c.channelID, CreatedAt: time.Now().UTC(), }, SequenceID: pkt.seqID, GatewayMessageID: fmt.Sprint(pkt.msgID), PhoneNumber: strings.TrimSpace(pkt.srcTerminalID), DestID: strings.TrimSpace(pkt.destID), Content: content, ReceivedAt: time.Now().UTC(), } if c.protocolLogPublisher != nil { c.emitProtocolLog(protocolLogEvent{Protocol: "cmpp", Direction: "channel_to_platform", EventType: "deliver_uplink", Status: "success", ChannelID: c.channelID, Account: c.config.Account, MessageID: cmd.MessageID, GatewayMessageID: fmt.Sprint(pkt.msgID), Phone: strings.TrimSpace(pkt.srcTerminalID), Detail: map[string]any{"sequenceId": pkt.seqID}}) } var publishErr error if c.eventPublisher != nil { publishErr = c.eventPublisher.PublishUplink(context.Background(), event) } else { publishErr = postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/uplink", event) } if err := publishErr; err != nil { log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_uplink status=forward_failed channel_id=%s sequence_id=%d packet_msg_id=%d error=%q", c.channelID, pkt.seqID, pkt.msgID, err) } else { log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_uplink status=forwarded channel_id=%s sequence_id=%d packet_msg_id=%d", c.channelID, pkt.seqID, pkt.msgID) } return nil } func (c *connection) decodeUplinkContent(pkt deliverPacket) (string, bool, error) { if pkt.tpUdhi != 1 { content, err := decodeContent(pkt.msgFmt, pkt.msgContent) return content, true, err } ref, total, number, payload, ok := parseConcatSegment(pkt.msgContent) if !ok { content, err := decodeContent(pkt.msgFmt, pkt.msgContent) return content, true, err } key := fmt.Sprintf("%s:%s:%s:%d:%d", c.channelID, strings.TrimSpace(pkt.srcTerminalID), strings.TrimSpace(pkt.destID), ref, total) c.mu.Lock() if c.longUplink == nil { c.longUplink = make(map[string]*longUplinkAssembly) } pruneLongUplinkAssemblies(c.longUplink, time.Now(), 10*time.Minute) content, complete, err := assembleLongUplink(c.longUplink, key, pkt.msgFmt, total, number, payload) c.mu.Unlock() return content, complete, err } func (c *connection) commandFor(gatewayMsgID uint64) (queue.SubmitCommand, bool) { c.mu.Lock() defer c.mu.Unlock() cmd, ok := c.tracker[gatewayMsgID] return cmd, ok } 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 receiptStatus(stat string) string { switch strings.ToUpper(strings.TrimSpace(stat)) { case "DELIVRD": return "delivered" case "": return "unknown" default: return "undelivered" } }