lislgosms/gateway/internal/inbound/server_test.go

package inbound

import (
	"bytes"
	"context"
	"encoding/binary"
	"encoding/json"
	"log"
	"net"
	"net/http"
	"net/http/httptest"
	"reflect"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	cmpp "github.com/bigwhite/gocmpp"
	cmpputils "github.com/bigwhite/gocmpp/utils"
)

type memoryPresenceStore struct {
	snapshots map[string]DownstreamPresence
	removed   []string
}

type memoryRecoveryStore struct {
	decisions map[string]RecoveryStartDecision
	completed []DownstreamRecoveryStatus
}

type synchronizedBuffer struct {
	mu     sync.Mutex
	buffer bytes.Buffer
}

func (b *synchronizedBuffer) Write(data []byte) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buffer.Write(data)
}

func (b *synchronizedBuffer) String() string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buffer.String()
}

func (m *memoryPresenceStore) TouchAccount(_ context.Context, snapshot DownstreamPresence) error {
	if m.snapshots == nil {
		m.snapshots = map[string]DownstreamPresence{}
	}
	m.snapshots[snapshot.Account] = snapshot
	return nil
}

func (m *memoryPresenceStore) RemoveAccount(_ context.Context, account string) error {
	delete(m.snapshots, account)
	m.removed = append(m.removed, account)
	return nil
}

func (m *memoryPresenceStore) ListAccounts(_ context.Context) ([]DownstreamPresence, error) {
	result := make([]DownstreamPresence, 0, len(m.snapshots))
	for _, snapshot := range m.snapshots {
		result = append(result, snapshot)
	}
	return result, nil
}

func (m *memoryRecoveryStore) StartAccountRecovery(_ context.Context, account string, _ string) (RecoveryStartDecision, error) {
	if decision, ok := m.decisions[account]; ok {
		return decision, nil
	}
	return RecoveryStartDecision{Allowed: true}, nil
}

func (m *memoryRecoveryStore) CompleteAccountRecovery(_ context.Context, status DownstreamRecoveryStatus) error {
	m.completed = append(m.completed, status)
	return nil
}

func (m *memoryRecoveryStore) ListRecoveryStatuses(_ context.Context) ([]DownstreamRecoveryStatus, error) {
	return append([]DownstreamRecoveryStatus(nil), m.completed...), nil
}

func (m *memoryRecoveryStore) GetAccountRecoveryStatus(_ context.Context, account string) (DownstreamRecoveryStatus, error) {
	for _, item := range m.completed {
		if item.Account == account {
			return item, nil
		}
	}
	return DownstreamRecoveryStatus{Account: account}, nil
}

func TestInboundServerAuthenticatesAndSubmits(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()
	account := "100001"
	password := "secret-hash"
	var gotAuth authRequest
	var gotSubmit submitRequest
	connectionEvents := make(chan downstreamConnectionEvent, 8)
	acknowledgements := make(chan map[string]any, 1)
	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/gateway/events/inbound/authenticate":
			if err := json.NewDecoder(r.Body).Decode(&gotAuth); err != nil {
				t.Fatalf("decode auth: %v", err)
			}
			_ = json.NewEncoder(w).Encode(authResponse{PasswordCipher: password, Account: account, EnterpriseCode: account})
		case "/api/gateway/events/inbound/submit":
			if err := json.NewDecoder(r.Body).Decode(&gotSubmit); err != nil {
				t.Fatalf("decode submit: %v", err)
			}
			_ = json.NewEncoder(w).Encode(submitResponse{
				Accepted: true, MessageID: "MSG-1",
				Messages: []submitResponseMessage{
					{PhoneNumber: "13500002696", MessageID: "MSG-1"},
					{PhoneNumber: "13600002696", MessageID: "MSG-2"},
				},
			})
		case "/api/gateway/events/downstream/pending":
			_ = json.NewEncoder(w).Encode([]pendingDelivery{})
		case "/api/gateway/events/inbound/connection":
			var event downstreamConnectionEvent
			if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
				t.Fatalf("decode connection event: %v", err)
			}
			connectionEvents <- event
			w.WriteHeader(http.StatusOK)
		case "/api/gateway/events/downstream/sent":
			w.WriteHeader(http.StatusOK)
		case "/api/gateway/events/downstream/acknowledged":
			var event map[string]any
			if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
				t.Fatalf("decode acknowledgement: %v", err)
			}
			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)
		}
	}))
	defer api.Close()

	addr := reserveTCPAddr(t)
	go func() {
		_ = (Server{Addr: addr, APIBaseURL: api.URL + "/api"}).ListenAndServe()
	}()
	time.Sleep(300 * time.Millisecond)

	client := cmpp.NewClient(cmpp.V30)
	defer client.Disconnect()
	if err := client.Connect(addr, account, password, 2*time.Second); err != nil {
		t.Fatalf("connect inbound cmpp: %v", err)
	}
	select {
	case event := <-connectionEvents:
		if event.Status != "connected" || event.Account != account || event.ConnectionID == "" || event.RemoteIP == "" || event.Protocol != "cmpp30" {
			t.Fatalf("unexpected connection event: %+v", event)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("expected downstream connected callback")
	}

	content, err := cmpputils.Utf8ToUcs2("测试入站")
	if err != nil {
		t.Fatalf("encode content: %v", err)
	}
	_, err = client.SendReqPkt(&cmpp.Cmpp3SubmitReqPkt{
		PkTotal:            1,
		PkNumber:           1,
		RegisteredDelivery: 1,
		MsgLevel:           1,
		ServiceId:          "cmpp",
		FeeUserType:        2,
		FeeTerminalId:      "13500002696",
		MsgFmt:             8,
		MsgSrc:             account,
		FeeType:            "02",
		FeeCode:            "0",
		SrcId:              "10690000",
		DestUsrTl:          2,
		DestTerminalId:     []string{"13500002696", "13600002696"},
		MsgLength:          uint8(len(content)),
		MsgContent:         content,
	})
	if err != nil {
		t.Fatalf("send submit: %v", err)
	}
	rsp := recvSubmitRsp(t, client)
	if rsp.Result != 0 || rsp.MsgId == 0 {
		t.Fatalf("unexpected submit response: %+v", rsp)
	}
	sendResult, err := PushReceiptWithResult(DownstreamReceipt{
		DeliveryID:    "delivery-1",
		MessageID:     "MSG-2",
		PhoneNumber:   "13600002696",
		ReceiptStatus: "delivered",
		DeliveredAt:   time.Now().UTC().Format(time.RFC3339Nano),
	})
	if err != nil || !sendResult.Sent {
		t.Fatalf("push receipt sent=%v err=%v", sendResult.Sent, err)
	}
	deliver := recvDeliver(t, client)
	if deliver.RegisterDelivery != 1 {
		t.Fatalf("expected receipt deliver, got %+v", deliver)
	}
	var receipt cmpp.CmppReceiptPkt
	if err := receipt.Unpack([]byte(deliver.MsgContent)); err != nil {
		t.Fatalf("unpack pushed receipt: %v", err)
	}
	if receipt.Stat != "DELIVRD" || receipt.DestTerminalId != "13600002696" || receipt.MsgId != rsp.MsgId {
		t.Fatalf("unexpected pushed receipt: %+v", receipt)
	}
	if err := client.SendRspPkt(&cmpp.Cmpp3DeliverRspPkt{MsgId: deliver.MsgId, Result: 0}, deliver.SeqId); err != nil {
		t.Fatalf("send deliver response: %v", err)
	}
	select {
	case event := <-acknowledgements:
		if event["id"] != "delivery-1" || event["result"] != float64(0) {
			t.Fatalf("unexpected acknowledgement callback: %+v", event)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("expected downstream acknowledgement callback")
	}
	if gotAuth.Account != account || gotAuth.AuthSource == "" || gotAuth.RemoteIP == "" {
		t.Fatalf("unexpected auth payload: %+v", gotAuth)
	}
	if gotSubmit.Account != account || gotSubmit.PhoneNumber != "13500002696" || gotSubmit.Content != "测试入站" ||
		!reflect.DeepEqual(gotSubmit.PhoneNumbers, []string{"13500002696", "13600002696"}) {
		t.Fatalf("unexpected submit payload: %+v", gotSubmit)
	}
}

func TestDecodeInboundLongMessageStripsConcatUDHBeforeUCS2Decode(t *testing.T) {
	payload, err := cmpputils.Utf8ToUcs2("【深圳市合正物业服务有限公司】第一片正文")
	if err != nil {
		t.Fatalf("encode content: %v", err)
	}
	raw := append([]byte{0x05, 0x00, 0x03, 0x10, 0x02, 0x01}, []byte(payload)...)

	content, fragment, err := decodeInboundSubmitContent(inboundSubmitPacket{
		pkTotal:    2,
		pkNumber:   1,
		tpUdhi:     1,
		msgFmt:     8,
		msgContent: string(raw),
	})
	if err != nil {
		t.Fatalf("decode long message: %v", err)
	}
	if content != "【深圳市合正物业服务有限公司】第一片正文" {
		t.Fatalf("decoded content = %q", content)
	}
	if fragment == nil || fragment.Reference != 0x10 || fragment.Total != 2 || fragment.Index != 1 {
		t.Fatalf("unexpected fragment metadata: %+v", fragment)
	}
}

func TestDecodeInboundLongMessageSupports16BitConcatReference(t *testing.T) {
	payload, err := cmpputils.Utf8ToUcs2("第二片正文")
	if err != nil {
		t.Fatalf("encode content: %v", err)
	}
	raw := append([]byte{0x06, 0x08, 0x04, 0x12, 0x34, 0x02, 0x02}, []byte(payload)...)

	content, fragment, err := decodeInboundSubmitContent(inboundSubmitPacket{
		pkTotal:    2,
		pkNumber:   2,
		tpUdhi:     1,
		msgFmt:     8,
		msgContent: string(raw),
	})
	if err != nil {
		t.Fatalf("decode long message: %v", err)
	}
	if content != "第二片正文" {
		t.Fatalf("decoded content = %q", content)
	}
	if fragment == nil || fragment.Reference != 0x1234 || fragment.Total != 2 || fragment.Index != 2 {
		t.Fatalf("unexpected fragment metadata: %+v", fragment)
	}
}

func TestInboundServerForwardsLongMessageFragmentsWithoutUDHAndAcknowledgesEachSubmit(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()
	account := "100001"
	password := "secret-hash"
	var mu sync.Mutex
	submits := make([]submitRequest, 0, 2)
	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/gateway/events/inbound/authenticate":
			_ = json.NewEncoder(w).Encode(authResponse{PasswordCipher: password, Account: account, EnterpriseCode: account})
		case "/api/gateway/events/inbound/submit":
			var submit submitRequest
			if err := json.NewDecoder(r.Body).Decode(&submit); err != nil {
				t.Fatalf("decode submit: %v", err)
			}
			mu.Lock()
			submits = append(submits, submit)
			count := len(submits)
			mu.Unlock()
			status := "fragment_pending"
			if count == 2 {
				status = "accepted"
			}
			_ = json.NewEncoder(w).Encode(map[string]any{
				"accepted": true, "messageId": "MSG-LONG-1", "status": status,
			})
		case "/api/gateway/events/downstream/pending":
			_ = 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)
		}
	}))
	defer api.Close()

	addr := reserveTCPAddr(t)
	go func() { _ = (Server{Addr: addr, APIBaseURL: api.URL + "/api"}).ListenAndServe() }()
	time.Sleep(300 * time.Millisecond)

	client := cmpp.NewClient(cmpp.V20)
	defer client.Disconnect()
	if err := client.Connect(addr, account, password, 2*time.Second); err != nil {
		t.Fatalf("connect CMPP2 inbound: %v", err)
	}
	parts := []string{"【签名】第一片", "第二片正文"}
	var firstResponseMsgID uint64
	for index, text := range parts {
		payload, _ := cmpputils.Utf8ToUcs2(text)
		raw := append([]byte{0x05, 0x00, 0x03, 0x22, 0x02, byte(index + 1)}, []byte(payload)...)
		if _, err := client.SendReqPkt(&cmpp.Cmpp2SubmitReqPkt{
			PkTotal: 2, PkNumber: uint8(index + 1), TpUdhi: 1, RegisteredDelivery: 1, MsgLevel: 1,
			ServiceId: "cmpp", FeeUserType: 2, FeeTerminalId: "13500002696",
			MsgFmt: 8, MsgSrc: account, FeeType: "02", FeeCode: "0", SrcId: "10690000",
			DestUsrTl: 1, DestTerminalId: []string{"13500002696"}, MsgLength: uint8(len(raw)), MsgContent: string(raw),
		}); err != nil {
			t.Fatalf("send long-message fragment %d: %v", index+1, err)
		}
		if rsp := recvSubmitRsp20(t, client); rsp.Result != 0 || rsp.MsgId == 0 {
			t.Fatalf("unexpected fragment %d response: %+v", index+1, rsp)
		} else if index == 0 {
			firstResponseMsgID = rsp.MsgId
		}
	}

	mu.Lock()
	defer mu.Unlock()
	if len(submits) != 2 {
		t.Fatalf("submit API calls = %d, want 2", len(submits))
	}
	for index, submit := range submits {
		if submit.Content != parts[index] {
			t.Fatalf("fragment %d content = %q", index+1, submit.Content)
		}
		if submit.LongMessage == nil || submit.LongMessage.Reference != 0x22 ||
			submit.LongMessage.Total != 2 || submit.LongMessage.Index != index+1 || submit.LongMessage.Format != 8 {
			t.Fatalf("fragment %d metadata = %+v", index+1, submit.LongMessage)
		}
	}
	downstreamRegistry.RLock()
	session := downstreamRegistry.byMessageID["MSG-LONG-1"]
	downstreamRegistry.RUnlock()
	if session == nil || session.gatewayMsgID != firstResponseMsgID {
		t.Fatalf("stored long-message Msg_Id = %v, want first fragment Msg_Id %v", session, firstResponseMsgID)
	}
}

func TestSubmitResponsePrecedesQueuedFailureReceipt(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()
	account := "100001"
	password := "secret-hash"
	var mu sync.Mutex
	var submit submitRequest
	pendingReturned := false

	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/gateway/events/inbound/authenticate":
			_ = json.NewEncoder(w).Encode(authResponse{PasswordCipher: password, Account: account, EnterpriseCode: account})
		case "/api/gateway/events/inbound/submit":
			mu.Lock()
			defer mu.Unlock()
			if err := json.NewDecoder(r.Body).Decode(&submit); err != nil {
				t.Fatalf("decode submit: %v", err)
			}
			_ = json.NewEncoder(w).Encode(submitResponse{Accepted: true, MessageID: "MSG-ORDER"})
		case "/api/gateway/events/downstream/pending":
			mu.Lock()
			defer mu.Unlock()
			if submit.SequenceID == 0 || pendingReturned {
				_ = json.NewEncoder(w).Encode([]pendingDelivery{})
				return
			}
			payload, _ := json.Marshal(DownstreamReceipt{
				Account: account, MessageID: "MSG-ORDER", PhoneNumber: "13500002696",
				ReceiptStatus: "undelivered", RawStatus: "REJECTD", SubmitSequenceID: submit.SequenceID,
			})
			pendingReturned = true
			_ = json.NewEncoder(w).Encode([]pendingDelivery{{
				ID: "delivery-order", DeliveryType: "receipt", Payload: payload, CreatedAt: time.Now().UTC(),
			}})
		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)
		}
	}))
	defer api.Close()

	addr := reserveTCPAddr(t)
	go func() { _ = (Server{Addr: addr, APIBaseURL: api.URL + "/api"}).ListenAndServe() }()
	time.Sleep(300 * time.Millisecond)

	client := cmpp.NewClient(cmpp.V20)
	defer client.Disconnect()
	if err := client.Connect(addr, account, password, 2*time.Second); err != nil {
		t.Fatalf("connect CMPP2 inbound: %v", err)
	}
	content, _ := cmpputils.Utf8ToUcs2("测试回执顺序")
	if _, err := client.SendReqPkt(&cmpp.Cmpp2SubmitReqPkt{
		PkTotal: 1, PkNumber: 1, RegisteredDelivery: 1, MsgLevel: 1,
		ServiceId: "cmpp", FeeUserType: 2, FeeTerminalId: "13500002696",
		MsgFmt: 8, MsgSrc: account, FeeType: "02", FeeCode: "0", SrcId: "10690000",
		DestUsrTl: 1, DestTerminalId: []string{"13500002696"}, MsgLength: uint8(len(content)), MsgContent: content,
	}); err != nil {
		t.Fatalf("send submit: %v", err)
	}

	first, err := client.RecvAndUnpackPkt(2 * time.Second)
	if err != nil {
		t.Fatalf("receive first packet: %v", err)
	}
	submitResponse, ok := first.(*cmpp.Cmpp2SubmitRspPkt)
	if !ok || submitResponse.Result != 0 || submitResponse.MsgId == 0 {
		t.Fatalf("first packet must be successful SUBMIT_RESP, got %T %+v", first, first)
	}
	deliver := recvDeliver20(t, client)
	if deliver.MsgId != submitResponse.MsgId {
		t.Fatalf("receipt Msg_Id=%d does not match SUBMIT_RESP Msg_Id=%d", deliver.MsgId, submitResponse.MsgId)
	}
}

func TestDailyLimitRejectsSubmitSynchronouslyWithoutPendingReceipt(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()
	account := "100001"
	password := "secret-hash"
	var pendingCalls atomic.Int32

	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/gateway/events/inbound/authenticate":
			_ = json.NewEncoder(w).Encode(authResponse{PasswordCipher: password, Account: account, EnterpriseCode: account})
		case "/api/gateway/events/inbound/submit":
			_ = json.NewEncoder(w).Encode(submitResponse{Accepted: false, Result: 8, MessageID: "MSG-LIMIT"})
		case "/api/gateway/events/downstream/pending":
			pendingCalls.Add(1)
			_ = 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)
		}
	}))
	defer api.Close()

	addr := reserveTCPAddr(t)
	go func() { _ = (Server{Addr: addr, APIBaseURL: api.URL + "/api"}).ListenAndServe() }()
	time.Sleep(300 * time.Millisecond)

	client := cmpp.NewClient(cmpp.V20)
	defer client.Disconnect()
	if err := client.Connect(addr, account, password, 2*time.Second); err != nil {
		t.Fatalf("connect CMPP2 inbound: %v", err)
	}
	time.Sleep(100 * time.Millisecond)
	pendingCallsAfterBind := pendingCalls.Load()
	content, _ := cmpputils.Utf8ToUcs2("日限额拒绝")
	if _, err := client.SendReqPkt(&cmpp.Cmpp2SubmitReqPkt{
		PkTotal: 1, PkNumber: 1, RegisteredDelivery: 1, MsgLevel: 1,
		ServiceId: "cmpp", FeeUserType: 2, FeeTerminalId: "13500002696",
		MsgFmt: 8, MsgSrc: account, FeeType: "02", FeeCode: "0", SrcId: "10690000",
		DestUsrTl: 1, DestTerminalId: []string{"13500002696"}, MsgLength: uint8(len(content)), MsgContent: content,
	}); err != nil {
		t.Fatalf("send submit: %v", err)
	}

	rsp := recvSubmitRsp20(t, client)
	if rsp.Result != 8 || rsp.MsgId != 0 {
		t.Fatalf("expected synchronous daily-limit result=8 without Msg_Id, got %+v", rsp)
	}
	time.Sleep(100 * time.Millisecond)
	if pendingCalls.Load() != pendingCallsAfterBind {
		t.Fatalf("daily-limit rejection must not add downstream receipt polling, before=%d after=%d", pendingCallsAfterBind, pendingCalls.Load())
	}
}

func TestReceiptWithoutOriginalSequenceIsUnrecoverable(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()

	result, err := PushReceiptWithResult(DownstreamReceipt{
		DeliveryID:    "delivery-history",
		Account:       "100001",
		MessageID:     "MSG-HISTORY",
		ReceiptStatus: "undelivered",
	})
	if err != nil {
		t.Fatalf("push receipt: %v", err)
	}
	if result.Sent || result.Retryable || result.ReasonCode != "MISSING_SUBMIT_SEQUENCE_ID" {
		t.Fatalf("unexpected result: %+v", result)
	}
}

func TestRecoverableReceiptWaitsForClientConnection(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()

	result, err := PushReceiptWithResult(DownstreamReceipt{
		DeliveryID:       "delivery-retry",
		Account:          "100001",
		MessageID:        "MSG-RETRY",
		SubmitSequenceID: 77,
		ReceiptStatus:    "delivered",
	})
	if err != nil {
		t.Fatalf("push receipt: %v", err)
	}
	if result.Sent || !result.Retryable || result.ReasonCode != "CLIENT_DISCONNECTED" {
		t.Fatalf("unexpected result: %+v", result)
	}
}

func TestInboundServerNegotiatesCMPP2AndUsesAuthenticatedAccountForSubmit(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()
	account := "100001"
	password := "secret-hash"
	var gotSubmit submitRequest
	submitCalls := 0
	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/gateway/events/inbound/authenticate":
			_ = json.NewEncoder(w).Encode(authResponse{PasswordCipher: password, Account: account, EnterpriseCode: "SP0001"})
		case "/api/gateway/events/inbound/submit":
			submitCalls++
			if err := json.NewDecoder(r.Body).Decode(&gotSubmit); err != nil {
				t.Fatalf("decode submit: %v", err)
			}
			_ = json.NewEncoder(w).Encode(submitResponse{Accepted: true, MessageID: "MSG-CMPP2"})
		case "/api/gateway/events/downstream/pending":
			_ = 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)
		}
	}))
	defer api.Close()

	addr := reserveTCPAddr(t)
	go func() {
		_ = (Server{Addr: addr, APIBaseURL: api.URL + "/api"}).ListenAndServe()
	}()
	time.Sleep(300 * time.Millisecond)

	client := cmpp.NewClient(cmpp.V20)
	defer client.Disconnect()
	if err := client.Connect(addr, account, password, 2*time.Second); err != nil {
		t.Fatalf("connect CMPP2 inbound: %v", err)
	}
	content, err := cmpputils.Utf8ToUcs2("测试CMPP2")
	if err != nil {
		t.Fatalf("encode content: %v", err)
	}
	_, err = client.SendReqPkt(&cmpp.Cmpp2SubmitReqPkt{
		PkTotal: 1, PkNumber: 1, RegisteredDelivery: 1, MsgLevel: 1,
		ServiceId: "cmpp", FeeUserType: 2, FeeTerminalId: "13500002696",
		MsgFmt: 8, MsgSrc: "SP0001", FeeType: "02", FeeCode: "0",
		SrcId: "10690000", DestUsrTl: 1, DestTerminalId: []string{"13500002696"},
		MsgLength: uint8(len(content)), MsgContent: content,
	})
	if err != nil {
		t.Fatalf("send CMPP2 submit: %v", err)
	}
	rsp := recvSubmitRsp20(t, client)
	if rsp.Result != 0 || rsp.MsgId == 0 {
		t.Fatalf("unexpected CMPP2 submit response: %+v", rsp)
	}
	if gotSubmit.Account != account || gotSubmit.PhoneNumber != "13500002696" || gotSubmit.Content != "测试CMPP2" {
		t.Fatalf("unexpected CMPP2 submit payload: %+v", gotSubmit)
	}
	_, err = client.SendReqPkt(&cmpp.Cmpp2SubmitReqPkt{
		PkTotal: 1, PkNumber: 1, RegisteredDelivery: 1, MsgLevel: 1,
		ServiceId: "cmpp", FeeUserType: 2, FeeTerminalId: "13500002696",
		MsgFmt: 8, MsgSrc: "BAD001", FeeType: "02", FeeCode: "0",
		SrcId: "10690000", DestUsrTl: 1, DestTerminalId: []string{"13500002696"},
		MsgLength: uint8(len(content)), MsgContent: content,
	})
	if err != nil {
		t.Fatalf("send mismatched enterprise code: %v", err)
	}
	if rejected := recvSubmitRsp20(t, client); rejected.Result != 9 {
		t.Fatalf("expected enterprise code rejection, got %+v", rejected)
	}
	if submitCalls != 1 {
		t.Fatalf("submit API calls = %d, want 1", submitCalls)
	}
	delivered, err := PushReceipt(DownstreamReceipt{
		MessageID: "MSG-CMPP2", PhoneNumber: "13500002696", ReceiptStatus: "delivered",
		DeliveredAt: time.Now().UTC().Format(time.RFC3339Nano),
	})
	if err != nil || !delivered {
		t.Fatalf("push CMPP2 receipt delivered=%v err=%v", delivered, err)
	}
	deliver := recvDeliver20(t, client)
	if deliver.RegisterDelivery != 1 {
		t.Fatalf("expected CMPP2 receipt deliver, got %+v", deliver)
	}
}

func TestPostIncludesAPIErrorResponseBody(t *testing.T) {
	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusBadRequest)
		_, _ = w.Write([]byte(`{"message":"CMPP submit content does not match an approved template and signature","statusCode":400}`))
	}))
	defer api.Close()

	err := (Server{APIBaseURL: api.URL}).post(context.Background(), "/inbound/submit", map[string]string{"account": "100001"}, nil)
	if err == nil || !bytes.Contains([]byte(err.Error()), []byte("CMPP submit content does not match an approved template and signature")) {
		t.Fatalf("expected API response body in error, got %v", err)
	}
}

func TestInboundServerLogsReadUnpackFailure(t *testing.T) {
	addr := reserveTCPAddr(t)
	var logs synchronizedBuffer
	go func() {
		_ = (Server{Addr: addr, LogWriter: &logs}).ListenAndServe()
	}()
	time.Sleep(300 * time.Millisecond)

	conn, err := net.DialTimeout("tcp", addr, time.Second)
	if err != nil {
		t.Fatalf("connect inbound server: %v", err)
	}
	defer conn.Close()
	if err := binary.Write(conn, binary.BigEndian, uint32(1)); err != nil {
		t.Fatalf("write invalid packet length: %v", err)
	}

	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if strings.Contains(logs.String(), "read/unpack packet failed") && strings.Contains(logs.String(), "total_length") {
			return
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatalf("missing read/unpack failure log: %s", logs.String())
}

func TestNormalizeInboundSubmitSupportsCMPP2AndCMPP3(t *testing.T) {
	tests := []struct {
		name     string
		packet   any
		protocol string
	}{
		{name: "cmpp2", packet: &cmpp.Cmpp2SubmitReqPkt{MsgSrc: "100001", SeqId: 20}, protocol: "cmpp20"},
		{name: "cmpp3", packet: &cmpp.Cmpp3SubmitReqPkt{MsgSrc: "100001", SeqId: 30}, protocol: "cmpp30"},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			got, ok := normalizeInboundSubmit(test.packet)
			if !ok || got.protocol != test.protocol || got.msgSrc != "100001" {
				t.Fatalf("unexpected normalized packet: %+v ok=%v", got, ok)
			}
		})
	}
}

func TestSetInboundSubmitResponseSupportsCMPP2AndCMPP3(t *testing.T) {
	cmpp2 := &cmpp.Cmpp2SubmitRspPkt{}
	setInboundSubmitResponse(cmpp2, 101, 9)
	if cmpp2.MsgId != 101 || cmpp2.Result != 9 {
		t.Fatalf("unexpected CMPP2 response: %+v", cmpp2)
	}
	cmpp3 := &cmpp.Cmpp3SubmitRspPkt{}
	setInboundSubmitResponse(cmpp3, 202, 9)
	if cmpp3.MsgId != 202 || cmpp3.Result != 9 {
		t.Fatalf("unexpected CMPP3 response: %+v", cmpp3)
	}
}

func TestFindSessionByConnUsesAuthenticatedConnection(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()
	conn := &cmpp.Conn{}
	session := &downstreamSession{
		account:  "100001",
		protocol: "cmpp20",
		conn:     conn,
	}
	downstreamRegistry.byConn[conn] = session
	if got := findSessionByConn(conn); got != session {
		t.Fatalf("unexpected session: %+v", got)
	}
	if got := findSessionByConn(&cmpp.Conn{}); got != nil {
		t.Fatalf("expected missing session, got %+v", got)
	}
}

func TestFlushOnlineAccountsFetchesPendingDeliveries(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()

	account := "100001"
	calls := 0
	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/gateway/events/downstream/pending":
			calls++
			var payload pendingDeliveryRequest
			if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
				t.Fatalf("decode pending request: %v", err)
			}
			if payload.Account != account {
				t.Fatalf("unexpected account: %+v", payload)
			}
			_ = json.NewEncoder(w).Encode([]pendingDelivery{})
		default:
			t.Fatalf("unexpected api path: %s", r.URL.Path)
		}
	}))
	defer api.Close()

	downstreamRegistry.Lock()
	downstreamRegistry.byAccount[account] = &downstreamSession{account: account}
	downstreamRegistry.Unlock()

	server := Server{APIBaseURL: api.URL + "/api"}
	server.flushOnlineAccounts(log.Default())

	if calls != 1 {
		t.Fatalf("pending fetch calls = %d, want 1", calls)
	}
}

func TestRecoverPendingCandidatesFetchesPresenceAccounts(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()

	account := "100009"
	calls := 0
	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/gateway/events/downstream/pending":
			calls++
			var payload pendingDeliveryRequest
			if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
				t.Fatalf("decode pending request: %v", err)
			}
			if payload.Account != account {
				t.Fatalf("unexpected account: %+v", payload)
			}
			_ = json.NewEncoder(w).Encode([]pendingDelivery{})
		default:
			t.Fatalf("unexpected api path: %s", r.URL.Path)
		}
	}))
	defer api.Close()

	store := &memoryPresenceStore{
		snapshots: map[string]DownstreamPresence{
			account: {
				Account:           account,
				GatewayInstanceID: "gateway-a",
				State:             "connected",
				UpdatedAt:         time.Now().UTC(),
			},
		},
	}

	server := Server{APIBaseURL: api.URL + "/api", PresenceStore: store}
	server.recoverPendingCandidates(log.Default())

	if calls != 1 {
		t.Fatalf("recovery pending fetch calls = %d, want 1", calls)
	}
}

func TestRecoverPendingCandidatesWritesWaitingConnectionStatus(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()

	account := "100010"
	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/gateway/events/downstream/pending":
			_ = json.NewEncoder(w).Encode([]pendingDelivery{{
				ID:           "delivery-1",
				DeliveryType: "receipt",
				Payload:      json.RawMessage(`{"account":"100010","messageId":"MSG-404","submitSequenceId":77,"phoneNumber":"13800000001","receiptStatus":"delivered"}`),
			}})
		case "/api/gateway/events/downstream/failed":
			w.WriteHeader(http.StatusOK)
		case "/api/gateway/events/downstream/recovery-status":
			w.WriteHeader(http.StatusOK)
		default:
			t.Fatalf("unexpected api path: %s", r.URL.Path)
		}
	}))
	defer api.Close()

	recovery := &memoryRecoveryStore{}
	store := &memoryPresenceStore{
		snapshots: map[string]DownstreamPresence{
			account: {
				Account:           account,
				GatewayInstanceID: "gateway-a",
				State:             "connected",
				UpdatedAt:         time.Now().UTC(),
			},
		},
	}

	server := Server{APIBaseURL: api.URL + "/api", PresenceStore: store, RecoveryStore: recovery}
	server.recoverPendingCandidates(log.Default())

	if len(recovery.completed) != 1 {
		t.Fatalf("completed recovery statuses = %d, want 1", len(recovery.completed))
	}
	if recovery.completed[0].State != "waiting_connection" {
		t.Fatalf("unexpected recovery status: %+v", recovery.completed[0])
	}
}

func TestRememberAndForgetAccountUpdatesPresenceStore(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()

	store := &memoryPresenceStore{}
	session := downstreamSession{
		account:     "100001",
		srcID:       "10690000",
		remoteIP:    "127.0.0.1",
		connectedAt: time.Now().UTC(),
		conn:        &cmpp.Conn{},
		mu:          &sync.Mutex{},
		presence:    store,
		instanceID:  "gateway-a",
	}

	if !rememberAccount(&session, 1) {
		t.Fatal("expected account session to be accepted")
	}
	snapshot, ok := store.snapshots["100001"]
	if !ok {
		t.Fatal("expected presence snapshot to be stored")
	}
	if snapshot.Account != "100001" || snapshot.GatewayInstanceID != "gateway-a" || snapshot.State != "connected" {
		t.Fatalf("unexpected snapshot: %+v", snapshot)
	}

	forgetDownstream(downstreamRegistry.byAccount["100001"])
	if len(store.removed) != 1 || store.removed[0] != "100001" {
		t.Fatalf("unexpected removed accounts: %+v", store.removed)
	}
}

func TestRememberAccountEnforcesConfiguredConnectionLimit(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()

	first := &downstreamSession{account: "100001", connectionID: "conn-1", conn: &cmpp.Conn{}, mu: &sync.Mutex{}}
	second := &downstreamSession{account: "100001", connectionID: "conn-2", conn: &cmpp.Conn{}, mu: &sync.Mutex{}}
	third := &downstreamSession{account: "100001", connectionID: "conn-3", conn: &cmpp.Conn{}, mu: &sync.Mutex{}}
	if !rememberAccount(first, 2) || !rememberAccount(second, 2) {
		t.Fatal("expected first two sessions to fit maxConnections=2")
	}
	if rememberAccount(third, 2) {
		t.Fatal("expected third session to be rejected by maxConnections=2")
	}
	forgetDownstream(first)
	if !rememberAccount(third, 2) {
		t.Fatal("expected a new session after a previous connection is released")
	}
}

func TestDownstreamDeliveryRequiresAcknowledgement(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()

	events := make(chan downstreamDeliveryLifecycleEvent, 1)
	protocolEvents := make(chan protocolLogEvent, 1)
	conn := &cmpp.Conn{}
	session := &downstreamSession{
		conn: conn, connectionID: "conn-1",
		deliveryReport: func(event downstreamDeliveryLifecycleEvent) { events <- event },
		tenantID:       "tenant-1", applicationID: "app-1", account: "607532",
		messageID: "MSG-LONG-1", phoneNumber: "18821203795",
		protocolLog: func(event protocolLogEvent) { protocolEvents <- event },
	}
	registerDownstreamAck(session, "delivery-1", 37, 9016479179509871733, time.Now().Add(time.Second))
	handleDownstreamAcknowledgement(conn, 37, 9016479179509871733, 0, log.Default())

	select {
	case event := <-events:
		if event.Kind != "acknowledged" || event.DeliveryID != "delivery-1" || event.Result != 0 || event.SequenceID != 37 {
			t.Fatalf("unexpected acknowledgement event: %+v", event)
		}
	case <-time.After(time.Second):
		t.Fatal("timed out waiting acknowledgement event")
	}
	select {
	case event := <-protocolEvents:
		if event.Protocol != "cmpp" || event.Direction != "client_to_platform" || event.EventType != "deliver_resp" {
			t.Fatalf("unexpected acknowledgement protocol event: %+v", event)
		}
		if event.MessageID != "MSG-LONG-1" || event.GatewayMessageID != "9016479179509871733" || event.ResultCode != "0" {
			t.Fatalf("unexpected acknowledgement identifiers: %+v", event)
		}
	case <-time.After(time.Second):
		t.Fatal("timed out waiting acknowledgement protocol event")
	}
}

func TestReceiptLookupDoesNotFallbackToAccountBeforeSubmitMappingExists(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()

	conn := &cmpp.Conn{}
	session := &downstreamSession{
		account: "100001", protocol: "cmpp20", conn: conn, mu: &sync.Mutex{}, connectionID: "conn-1",
	}
	if !rememberAccount(session, 1) {
		t.Fatal("expected account session to be accepted")
	}
	if session := findReceiptSession("MSG-NOT-REMEMBERED", "100001"); session != nil {
		t.Fatalf("receipt unexpectedly fell back to account session: %+v", session)
	}

	recovered := recoverReceiptSession(DownstreamReceipt{
		MessageID: "MSG-NOT-REMEMBERED", Account: "100001", SubmitSequenceID: 1216579149,
	})
	if recovered == nil {
		t.Fatal("expected persisted submit sequence to recover receipt session")
	}
	if recovered.gatewayMsgID != messageIDFrom("MSG-NOT-REMEMBERED", 1216579149) || recovered.gatewayMsgID == 0 {
		t.Fatalf("unexpected recovered Msg_Id: %d", recovered.gatewayMsgID)
	}
	first := recoverReceiptSession(DownstreamReceipt{
		MessageID: "MSG-FIRST", SubmitGroupMessageID: "MSG-GROUP", Account: "100001", SubmitSequenceID: 77,
	})
	second := recoverReceiptSession(DownstreamReceipt{
		MessageID: "MSG-SECOND", SubmitGroupMessageID: "MSG-GROUP", Account: "100001", SubmitSequenceID: 77,
	})
	if first == nil || second == nil || first.gatewayMsgID != second.gatewayMsgID || first.gatewayMsgID != messageIDFrom("MSG-GROUP", 77) {
		t.Fatalf("multi-destination recovery did not preserve the original Msg_Id: first=%+v second=%+v", first, second)
	}
}

func TestSendDownstreamRejectsZeroMessageID(t *testing.T) {
	_, err := sendDownstream(
		&downstreamSession{mu: &sync.Mutex{}},
		&cmpp.Cmpp2DeliverReqPkt{MsgId: 0},
		"delivery-zero",
	)
	if err == nil || !strings.Contains(err.Error(), "Msg_Id=0") {
		t.Fatalf("expected zero Msg_Id rejection, got %v", err)
	}
}

func TestDownstreamDeliveryReportsAckTimeout(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()

	events := make(chan downstreamDeliveryLifecycleEvent, 1)
	session := &downstreamSession{
		conn: &cmpp.Conn{}, connectionID: "conn-1",
		deliveryReport: func(event downstreamDeliveryLifecycleEvent) { events <- event },
	}
	registerDownstreamAck(session, "delivery-timeout", 38, 9017467844344255865, time.Now().Add(20*time.Millisecond))

	select {
	case event := <-events:
		if event.Kind != "failed" || event.FailureType != "ack_timeout" || event.DeliveryID != "delivery-timeout" {
			t.Fatalf("unexpected timeout event: %+v", event)
		}
	case <-time.After(time.Second):
		t.Fatal("timed out waiting acknowledgement timeout")
	}
}

func recvDeliver(t *testing.T, client *cmpp.Client) *cmpp.Cmpp3DeliverReqPkt {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		packet, err := client.RecvAndUnpackPkt(200 * time.Millisecond)
		if err != nil {
			continue
		}
		if deliver, ok := packet.(*cmpp.Cmpp3DeliverReqPkt); ok {
			return deliver
		}
	}
	t.Fatal("timed out waiting deliver request")
	return nil
}

func recvDeliver20(t *testing.T, client *cmpp.Client) *cmpp.Cmpp2DeliverReqPkt {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		packet, err := client.RecvAndUnpackPkt(200 * time.Millisecond)
		if err != nil {
			continue
		}
		if deliver, ok := packet.(*cmpp.Cmpp2DeliverReqPkt); ok {
			return deliver
		}
	}
	t.Fatal("timed out waiting CMPP2 deliver request")
	return nil
}

func resetDownstreamRegistry() {
	downstreamRegistry.Lock()
	defer downstreamRegistry.Unlock()
	downstreamRegistry.byAccount = make(map[string]*downstreamSession)
	downstreamRegistry.byMessageID = make(map[string]*downstreamSession)
	downstreamRegistry.byConn = make(map[*cmpp.Conn]*downstreamSession)
	downstreamAckRegistry.Lock()
	for _, tracker := range downstreamAckRegistry.items {
		if tracker.timer != nil {
			tracker.timer.Stop()
		}
	}
	downstreamAckRegistry.items = make(map[string]*downstreamAckTracker)
	downstreamAckRegistry.Unlock()
}

func reserveTCPAddr(t *testing.T) string {
	t.Helper()
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("reserve tcp addr: %v", err)
	}
	addr := listener.Addr().String()
	if err := listener.Close(); err != nil {
		t.Fatalf("close reserved listener: %v", err)
	}
	return addr
}

func recvSubmitRsp(t *testing.T, client *cmpp.Client) *cmpp.Cmpp3SubmitRspPkt {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		packet, err := client.RecvAndUnpackPkt(200 * time.Millisecond)
		if err != nil {
			continue
		}
		if rsp, ok := packet.(*cmpp.Cmpp3SubmitRspPkt); ok {
			return rsp
		}
	}
	t.Fatal("timed out waiting submit response")
	return nil
}

func recvSubmitRsp20(t *testing.T, client *cmpp.Client) *cmpp.Cmpp2SubmitRspPkt {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		packet, err := client.RecvAndUnpackPkt(200 * time.Millisecond)
		if err != nil {
			continue
		}
		if rsp, ok := packet.(*cmpp.Cmpp2SubmitRspPkt); ok {
			return rsp
		}
	}
	t.Fatal("timed out waiting CMPP2 submit response")
	return nil
}