lislgosms/gateway/internal/inbound/server_test.go

package inbound

import (
	"bytes"
	"context"
	"encoding/json"
	"log"
	"net"
	"net/http"
	"net/http/httptest"
	"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
}

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) {
	account := "100001"
	password := "secret-hash"
	var gotAuth authRequest
	var gotSubmit submitRequest
	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})
		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{})
		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)
	}

	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)
	}
	delivered, err := PushReceipt(DownstreamReceipt{
		MessageID:     "MSG-1",
		PhoneNumber:   "13500002696",
		ReceiptStatus: "delivered",
		DeliveredAt:   time.Now().UTC().Format(time.RFC3339Nano),
	})
	if err != nil || !delivered {
		t.Fatalf("push receipt delivered=%v err=%v", delivered, 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 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 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 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 TestInboundClientProtocolUsesConnectRequestVersion(t *testing.T) {
	resetDownstreamRegistry()
	defer resetDownstreamRegistry()
	conn := &cmpp.Conn{}
	downstreamRegistry.byAccount["100001"] = &downstreamSession{
		account:  "100001",
		protocol: "cmpp20",
		conn:     conn,
	}
	if got := inboundClientProtocol("100001", conn, "cmpp30"); got != "cmpp20" {
		t.Fatalf("protocol = %s, want cmpp20", got)
	}
	if got := inboundClientProtocol("missing", conn, "cmpp30"); got != "cmpp30" {
		t.Fatalf("fallback protocol = %s, want cmpp30", 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 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 resetDownstreamRegistry() {
	downstreamRegistry.Lock()
	defer downstreamRegistry.Unlock()
	downstreamRegistry.byAccount = make(map[string]*downstreamSession)
	downstreamRegistry.byMessageID = make(map[string]*downstreamSession)
}

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
}