feat: densify sms records and improve uplink matching

gateway/internal/queue/messages.go

@@ -19,7 +19,7 @@ type Envelope struct {
 	SchemaVersion string      `json:"schemaVersion"`
 	MessageType   MessageType `json:"messageType"`
 	TraceID       string      `json:"traceId"`
-	MessageID     string      `json:"messageId"`
+	MessageID     string      `json:"messageId,omitempty"`
 	ChannelID     string      `json:"channelId"`
 	CreatedAt     time.Time   `json:"createdAt"`
 }
@@ -118,11 +118,12 @@ type ReceiptEvent struct {
 
 type UplinkEvent struct {
 	Envelope
-	SequenceID  uint32    `json:"sequenceId"`
-	PhoneNumber string    `json:"phoneNumber"`
-	DestID      string    `json:"destId"`
-	Content     string    `json:"content"`
-	ReceivedAt  time.Time `json:"receivedAt"`
+	SequenceID       uint32    `json:"sequenceId"`
+	GatewayMessageID string    `json:"gatewayMessageId"`
+	PhoneNumber      string    `json:"phoneNumber"`
+	DestID           string    `json:"destId"`
+	Content          string    `json:"content"`
+	ReceivedAt       time.Time `json:"receivedAt"`
 }
 
 type ConnectChannelCommand struct {

gateway/internal/upstream/deliver.go

@@ -124,14 +124,15 @@ func (c *connection) handleDeliver(pkt deliverPacket) error {
 			ChannelID:     c.channelID,
 			CreatedAt:     time.Now().UTC(),
 		},
-		SequenceID:  pkt.seqID,
-		PhoneNumber: strings.TrimSpace(pkt.srcTerminalID),
-		DestID:      strings.TrimSpace(pkt.destID),
-		Content:     content,
-		ReceivedAt:  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, Phone: strings.TrimSpace(pkt.srcTerminalID), Detail: map[string]any{"sequenceId": pkt.seqID}})
+		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 {

gateway/internal/upstream/deliver_test.go

@@ -92,6 +92,54 @@ func TestHandleCMPP2DeliverReceiptPostsReceiptEvent(t *testing.T) {
 	}
 }
 
+func TestHandleCMPP2UplinkPreservesGatewayMessageIDWithoutPretendingItIsASubmitMessage(t *testing.T) {
+	events := make(chan queue.UplinkEvent, 1)
+	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/gateway/events/uplink" {
+			t.Fatalf("unexpected path: %s", r.URL.Path)
+		}
+		var event queue.UplinkEvent
+		if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
+			t.Fatalf("decode uplink event: %v", err)
+		}
+		events <- event
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer api.Close()
+
+	conn := &connection{
+		channelID:  "channel-1",
+		apiBaseURL: api.URL,
+		httpClient: api.Client(),
+		tracker:    map[uint64]queue.SubmitCommand{},
+	}
+	if err := conn.handleDeliver(deliverPacketFromCMPP2(&cmpp.Cmpp2DeliverReqPkt{
+		SeqId:            8,
+		MsgId:            8412634832294102675,
+		DestId:           "10690000",
+		SrcTerminalId:    "13800000001",
+		RegisterDelivery: 0,
+		MsgContent:       "TD",
+	})); err != nil {
+		t.Fatalf("handle uplink: %v", err)
+	}
+
+	select {
+	case event := <-events:
+		if event.GatewayMessageID != "8412634832294102675" {
+			t.Fatalf("GatewayMessageID = %q", event.GatewayMessageID)
+		}
+		if event.MessageID != "" {
+			t.Fatalf("MessageID = %q, want empty without a correlated submit", event.MessageID)
+		}
+		if event.PhoneNumber != "13800000001" || event.DestID != "10690000" || event.Content != "TD" {
+			t.Fatalf("unexpected uplink event: %+v", event)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for uplink event")
+	}
+}
+
 func TestReceiptStatusTreatsNonDeliveredFinalStatesAsUndelivered(t *testing.T) {
 	for _, stat := range []string{"UNKNOWN", "UNDELIV", "EXPIRED", "DELETED", "REJECTD"} {
 		if got := receiptStatus(stat); got != "undelivered" {