package inbound

import (
	"bytes"
	"context"
	"encoding/binary"
	"encoding/json"
	"log"
	"net"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"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"})
		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)
		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:          1,
		DestTerminalId:     []string{"13500002696"},
		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-1",
		PhoneNumber:   "13500002696",
		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 != "13500002696" {
		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 != "测试入站" {
		t.Fatalf("unexpected submit payload: %+v", gotSubmit)
	}
}

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)
		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(`{"messageId":"MSG-404","phoneNumber":"13800000001","receiptStatus":"delivered"}`),
			}})
		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",
	}

	rememberAccount(session)
	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 TestDownstreamDeliveryRequiresAcknowledgement(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()

	events := make(chan downstreamDeliveryLifecycleEvent, 1)
	conn := &cmpp.Conn{}
	session := &downstreamSession{
		conn: conn, connectionID: "conn-1",
		deliveryReport: func(event downstreamDeliveryLifecycleEvent) { events <- 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")
	}
}

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
}