lislgosms/gateway/internal/upstream/manager.go

package upstream

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"net/http"
	"strings"
	"sync"
	"time"

	"cmpp-platform/gateway/internal/queue"

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

const (
	defaultConnectTimeout = 5 * time.Second
	defaultSubmitTimeout  = 10 * time.Second
	defaultHTTPTimeout    = 10 * time.Second
	defaultWindowSize     = 16
)

type Manager struct {
	APIBaseURL string
	HTTPClient *http.Client

	mu    sync.Mutex
	conns map[string]*connectionPool
}

type ConnectionState struct {
	ChannelID          string `json:"channelId"`
	ConnectionID       string `json:"connectionId"`
	Status             string `json:"status"`
	DesiredConnections int    `json:"desiredConnections"`
	CurrentConnections int    `json:"currentConnections"`
	LastConnectedAt    string `json:"lastConnectedAt,omitempty"`
	LastDisconnectedAt string `json:"lastDisconnectedAt,omitempty"`
	LastHeartbeatAt    string `json:"lastHeartbeatAt,omitempty"`
	ReconnectCount     int    `json:"reconnectCount,omitempty"`
	LastError          string `json:"lastError,omitempty"`
}

func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, error) {
	if err := validateSubmitCommand(cmd); err != nil {
		result := submitResult(cmd, 0, "", "rejected", "INVALID_COMMAND", err.Error())
		if postErr := m.post(ctx, "/gateway/events/submit-result", result); postErr != nil {
			return result, postErr
		}
		return result, err
	}

	pool, err := m.connectionFor(cmd)
	if err != nil {
		result := submitResult(cmd, 0, "", "rejected", "CONNECT_FAILED", err.Error())
		if postErr := m.post(ctx, "/gateway/events/submit-result", result); postErr != nil {
			return result, postErr
		}
		return result, err
	}

	result, err := pool.submit(ctx, cmd)
	if err != nil {
		if postErr := m.post(ctx, "/gateway/events/submit-result", result); postErr != nil {
			return result, postErr
		}
		return result, err
	}
	if err := m.post(ctx, "/gateway/events/submit-result", result); err != nil {
		return result, err
	}
	return result, nil
}

func (m *Manager) ConnectChannel(ctx context.Context, command queue.ConnectChannelCommand) (ConnectionState, error) {
	if err := validateConnectChannelCommand(command); err != nil {
		return ConnectionState{}, err
	}

	m.mu.Lock()
	m.ensureDefaultsLocked()
	pool := m.conns[command.ChannelID]
	config := normalizeUpstreamConfig(queue.UpstreamConfig{
		GatewayHost:        command.Channel.GatewayHost,
		GatewayPort:        command.Channel.GatewayPort,
		Account:            command.Channel.Account,
		PasswordCipher:     command.Channel.PasswordCipher,
		CMPPVersion:        command.Channel.CMPPVersion,
		DesiredConnections: command.DesiredConnections,
		WindowSize:         16,
	})
	if pool == nil || !pool.matches(config) {
		if pool != nil {
			pool.close()
		}
		pool = m.newConnectionPool(command.ChannelID, command.ConnectionID, config)
		m.conns[command.ChannelID] = pool
	}
	m.mu.Unlock()

	if err := pool.ensureConnected(); err != nil {
		_ = pool.reportState(ctx, "failed", err)
		m.mu.Lock()
		delete(m.conns, command.ChannelID)
		m.mu.Unlock()
		return pool.snapshotState("failed", err), nil
	}
	return pool.snapshotState("connected", nil), nil
}

func (m *Manager) connectionFor(cmd queue.SubmitCommand) (*connectionPool, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.ensureDefaultsLocked()

	pool := m.conns[cmd.ChannelID]
	if pool == nil || !pool.matches(cmd.Upstream) {
		if pool != nil {
			pool.close()
		}
		pool = m.newConnectionPool(cmd.ChannelID, defaultChannelConnectionID(cmd.ChannelID), normalizeUpstreamConfig(cmd.Upstream))
		m.conns[cmd.ChannelID] = pool
	}
	if err := pool.ensureConnected(); err != nil {
		delete(m.conns, cmd.ChannelID)
		return nil, err
	}
	return pool, nil
}

func (m *Manager) post(ctx context.Context, path string, payload any) error {
	client := m.HTTPClient
	if client == nil {
		client = &http.Client{Timeout: defaultHTTPTimeout}
	}
	return postJSON(ctx, client, m.APIBaseURL, path, payload)
}

func (m *Manager) ensureDefaultsLocked() {
	if m.HTTPClient == nil {
		m.HTTPClient = &http.Client{Timeout: defaultHTTPTimeout}
	}
	if m.conns == nil {
		m.conns = make(map[string]*connectionPool)
	}
}

func (m *Manager) newConnectionPool(channelID string, connectionID string, config queue.UpstreamConfig) *connectionPool {
	return &connectionPool{
		channelID:    channelID,
		connectionID: connectionID,
		config:       config,
		apiBaseURL:   m.APIBaseURL,
		httpClient:   m.HTTPClient,
		reporter: func(ctx context.Context, state ConnectionState) error {
			return m.post(ctx, "/admin/gateway/connections", state)
		},
	}
}

type connectionPool struct {
	channelID    string
	connectionID string
	config       queue.UpstreamConfig
	apiBaseURL   string
	httpClient   *http.Client
	reporter     func(context.Context, ConnectionState) error

	mu    sync.Mutex
	conns []*connection
	next  int
}

func (p *connectionPool) matches(config queue.UpstreamConfig) bool {
	return p.config == normalizeUpstreamConfig(config)
}

func (p *connectionPool) ensureConnected() error {
	desired := p.config.DesiredConnections
	if desired <= 0 {
		desired = 1
	}
	connectedAny := false
	for {
		p.mu.Lock()
		if len(p.conns) >= desired {
			p.mu.Unlock()
			break
		}
		index := len(p.conns)
		conn := &connection{
			channelID:  p.channelID,
			config:     p.config,
			index:      index,
			pool:       p,
			apiBaseURL: p.apiBaseURL,
			httpClient: p.httpClient,
			window:     make(chan struct{}, p.config.WindowSize),
			pending:    make(map[uint32]chan submitPartResponse),
			tracker:    make(map[uint64]queue.SubmitCommand),
			longUplink: make(map[string]*longUplinkAssembly),
		}
		p.mu.Unlock()

		connected, err := conn.ensureConnected()
		if err != nil {
			conn.close()
			p.close()
			return err
		}

		p.mu.Lock()
		p.conns = append(p.conns, conn)
		p.mu.Unlock()
		connectedAny = connectedAny || connected
	}
	if connectedAny {
		_ = p.reportState(context.Background(), "connected", nil)
	}
	return nil
}

func (p *connectionPool) submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, error) {
	parts, err := splitSubmitContent(cmd.CMPP.MsgFmt, cmd.Content)
	if err != nil {
		result := submitResult(cmd, 0, "", "rejected", "ENCODE_FAILED", err.Error())
		return result, err
	}

	var firstSequence uint32
	var firstGatewayMessageID string
	segments := make([]queue.SubmitSegmentResult, 0, len(parts))
	for _, part := range parts {
		conn, release, err := p.acquireConnection(ctx)
		if err != nil {
			result := submitResult(cmd, 0, "", "timeout", "WINDOW_TIMEOUT", err.Error())
			result.Segments = segments
			return result, err
		}
		seq, gatewayMessageID, result, err := conn.submitPart(ctx, cmd, part)
		release()
		segments = append(segments, submitSegmentResult(part, seq, gatewayMessageID, result))
		if firstSequence == 0 {
			firstSequence = seq
		}
		if firstGatewayMessageID == "" {
			firstGatewayMessageID = gatewayMessageID
		}
		if err != nil {
			result.Segments = segments
			return result, err
		}
		if result.SubmitStatus != "accepted" {
			result.Segments = segments
			return result, nil
		}
	}

	result := submitResult(cmd, firstSequence, firstGatewayMessageID, "accepted", "", "")
	result.Segments = segments
	return result, nil
}

func (p *connectionPool) acquireConnection(ctx context.Context) (*connection, func(), error) {
	waitCtx, cancel := context.WithTimeout(ctx, defaultSubmitTimeout)
	defer cancel()
	ticker := time.NewTicker(10 * time.Millisecond)
	defer ticker.Stop()

	for {
		if conn, release := p.tryAcquireConnection(); conn != nil {
			if connected, err := conn.ensureConnected(); err != nil {
				release()
				select {
				case <-waitCtx.Done():
					return nil, nil, waitCtx.Err()
				case <-ticker.C:
					continue
				}
			} else if connected {
				_ = p.reportState(context.Background(), "connected", nil)
			}
			return conn, release, nil
		}
		select {
		case <-waitCtx.Done():
			return nil, nil, waitCtx.Err()
		case <-ticker.C:
		}
	}
}

func (p *connectionPool) tryAcquireConnection() (*connection, func()) {
	p.mu.Lock()
	defer p.mu.Unlock()
	if len(p.conns) == 0 {
		return nil, nil
	}
	for i := 0; i < len(p.conns); i++ {
		index := (p.next + i) % len(p.conns)
		conn := p.conns[index]
		if conn.tryAcquireWindow() {
			p.next = (index + 1) % len(p.conns)
			return conn, conn.releaseWindow
		}
	}
	return nil, nil
}

func (p *connectionPool) close() {
	p.mu.Lock()
	conns := p.conns
	p.conns = nil
	p.mu.Unlock()
	for _, conn := range conns {
		conn.close()
	}
}

func (p *connectionPool) reportState(ctx context.Context, status string, stateErr error) error {
	if p.reporter == nil {
		return nil
	}
	return p.reporter(ctx, p.snapshotState(status, stateErr))
}

func (p *connectionPool) snapshotState(status string, stateErr error) ConnectionState {
	now := time.Now().UTC().Format(time.RFC3339Nano)
	state := ConnectionState{
		ChannelID:          p.channelID,
		ConnectionID:       p.connectionID,
		Status:             status,
		DesiredConnections: p.config.DesiredConnections,
		CurrentConnections: p.countActiveConnections(),
	}
	if state.DesiredConnections <= 0 {
		state.DesiredConnections = 1
	}
	switch status {
	case "connected":
		state.LastConnectedAt = now
		state.LastHeartbeatAt = now
	case "disconnected", "failed":
		state.LastDisconnectedAt = now
	}
	if stateErr != nil {
		state.LastError = stateErr.Error()
	}
	return state
}

func (p *connectionPool) countActiveConnections() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	count := 0
	for _, conn := range p.conns {
		conn.mu.Lock()
		active := conn.client != nil && !conn.closed
		conn.mu.Unlock()
		if active {
			count += 1
		}
	}
	return count
}

type connection struct {
	channelID  string
	config     queue.UpstreamConfig
	index      int
	pool       *connectionPool
	apiBaseURL string
	httpClient *http.Client

	mu         sync.Mutex
	sendMu     sync.Mutex
	client     *cmpp.Client
	window     chan struct{}
	pending    map[uint32]chan submitPartResponse
	tracker    map[uint64]queue.SubmitCommand
	longUplink map[string]*longUplinkAssembly
	readOnce   sync.Once
	closed     bool
}

type submitPartResponse struct {
	seqID  uint32
	msgID  uint64
	result uint32
	err    error
}

func (c *connection) matches(config queue.UpstreamConfig) bool {
	return c.config == normalizeUpstreamConfig(config)
}

func (c *connection) ensureConnected() (bool, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.client != nil && !c.closed {
		return false, nil
	}
	client := cmpp.NewClient(protocolVersion(c.config.CMPPVersion))
	addr := fmt.Sprintf("%s:%d", c.config.GatewayHost, c.config.GatewayPort)
	if err := client.Connect(addr, c.config.Account, c.config.PasswordCipher, defaultConnectTimeout); err != nil {
		client.Disconnect()
		return false, err
	}
	c.client = client
	c.closed = false
	go c.readLoop()
	return true, nil
}

func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, part submitPart) (uint32, string, queue.SubmitResult, error) {
	rspCh := make(chan submitPartResponse, 1)
	pkt := c.submitRequestPacket(cmd, part)

	c.sendMu.Lock()
	seq, err := c.client.SendReqPkt(pkt)
	c.sendMu.Unlock()
	if err != nil {
		c.close()
		result := submitResult(cmd, 0, "", "timeout", "SEND_FAILED", err.Error())
		return 0, "", result, err
	}

	c.mu.Lock()
	c.pending[seq] = rspCh
	c.mu.Unlock()
	defer func() {
		c.mu.Lock()
		delete(c.pending, seq)
		c.mu.Unlock()
	}()

	waitCtx, cancel := context.WithTimeout(ctx, defaultSubmitTimeout)
	defer cancel()
	select {
	case <-waitCtx.Done():
		result := submitResult(cmd, seq, "", "timeout", "SUBMIT_TIMEOUT", waitCtx.Err().Error())
		return seq, "", result, waitCtx.Err()
	case rsp := <-rspCh:
		if rsp.err != nil {
			result := submitResult(cmd, seq, "", "timeout", "CONNECTION_LOST", rsp.err.Error())
			return seq, "", result, rsp.err
		}
		gatewayMessageID := fmt.Sprint(rsp.msgID)
		status := "accepted"
		errorCode := ""
		errorMessage := ""
		if rsp.result != 0 {
			status = "rejected"
			errorCode = fmt.Sprint(rsp.result)
			errorMessage = fmt.Sprintf("upstream submit rejected with result %d", rsp.result)
		}
		if rsp.result == 0 {
			c.mu.Lock()
			c.tracker[rsp.msgID] = cmd
			c.mu.Unlock()
		}
		return seq, gatewayMessageID, submitResult(cmd, seq, gatewayMessageID, status, errorCode, errorMessage), nil
	}
}

func (c *connection) submitRequestPacket(cmd queue.SubmitCommand, part submitPart) cmpp.Packer {
	base := submitRequestFields{
		PkTotal:            part.PkTotal,
		PkNumber:           part.PkNumber,
		TpUdhi:             part.TpUdhi,
		RegisteredDelivery: uint8(cmd.CMPP.RegisteredDelivery),
		MsgLevel:           1,
		ServiceId:          cmd.CMPP.ServiceID,
		FeeUserType:        uint8(defaultInt(cmd.CMPP.FeeUserType, 2)),
		FeeTerminalId:      cmd.PhoneNumber,
		MsgFmt:             uint8(cmd.CMPP.MsgFmt),
		MsgSrc:             c.config.Account,
		FeeType:            defaultString(cmd.CMPP.FeeType, "02"),
		FeeCode:            defaultString(cmd.CMPP.FeeCode, "0"),
		SrcId:              cmd.CMPP.SrcID,
		DestUsrTl:          1,
		DestTerminalId:     []string{cmd.PhoneNumber},
		MsgLength:          uint8(len(part.MsgContent)),
		MsgContent:         part.MsgContent,
	}
	if protocolVersion(c.config.CMPPVersion) == cmpp.V20 {
		return &cmpp.Cmpp2SubmitReqPkt{
			PkTotal:            base.PkTotal,
			PkNumber:           base.PkNumber,
			RegisteredDelivery: base.RegisteredDelivery,
			MsgLevel:           base.MsgLevel,
			ServiceId:          base.ServiceId,
			FeeUserType:        base.FeeUserType,
			FeeTerminalId:      base.FeeTerminalId,
			TpUdhi:             base.TpUdhi,
			MsgFmt:             base.MsgFmt,
			MsgSrc:             base.MsgSrc,
			FeeType:            base.FeeType,
			FeeCode:            base.FeeCode,
			SrcId:              base.SrcId,
			DestUsrTl:          base.DestUsrTl,
			DestTerminalId:     base.DestTerminalId,
			MsgLength:          base.MsgLength,
			MsgContent:         base.MsgContent,
		}
	}
	return &cmpp.Cmpp3SubmitReqPkt{
		PkTotal:            base.PkTotal,
		PkNumber:           base.PkNumber,
		RegisteredDelivery: base.RegisteredDelivery,
		MsgLevel:           base.MsgLevel,
		ServiceId:          base.ServiceId,
		FeeUserType:        base.FeeUserType,
		FeeTerminalId:      base.FeeTerminalId,
		TpUdhi:             base.TpUdhi,
		MsgFmt:             base.MsgFmt,
		MsgSrc:             base.MsgSrc,
		FeeType:            base.FeeType,
		FeeCode:            base.FeeCode,
		SrcId:              base.SrcId,
		DestUsrTl:          base.DestUsrTl,
		DestTerminalId:     base.DestTerminalId,
		MsgLength:          base.MsgLength,
		MsgContent:         base.MsgContent,
	}
}

type submitRequestFields struct {
	PkTotal            uint8
	PkNumber           uint8
	RegisteredDelivery uint8
	MsgLevel           uint8
	ServiceId          string
	FeeUserType        uint8
	FeeTerminalId      string
	TpUdhi             uint8
	MsgFmt             uint8
	MsgSrc             string
	FeeType            string
	FeeCode            string
	SrcId              string
	DestUsrTl          uint8
	DestTerminalId     []string
	MsgLength          uint8
	MsgContent         string
}

func (c *connection) tryAcquireWindow() bool {
	if c.window == nil {
		c.window = make(chan struct{}, defaultWindowSize)
	}
	select {
	case c.window <- struct{}{}:
		return true
	default:
		return false
	}
}

func (c *connection) releaseWindow() {
	if c.window == nil {
		return
	}
	select {
	case <-c.window:
	default:
	}
}

func (c *connection) readLoop() {
	for {
		pkt, err := c.client.RecvAndUnpackPkt(time.Second)
		if err != nil {
			c.mu.Lock()
			closed := c.closed
			c.mu.Unlock()
			if closed {
				return
			}
			if isTemporaryReadTimeout(err) {
				continue
			}
			c.handleConnectionLoss(err)
			continue
		}
		switch p := pkt.(type) {
		case *cmpp.Cmpp2SubmitRspPkt:
			c.mu.Lock()
			ch := c.pending[p.SeqId]
			c.mu.Unlock()
			if ch != nil {
				ch <- submitPartResponse{seqID: p.SeqId, msgID: p.MsgId, result: uint32(p.Result)}
			}
		case *cmpp.Cmpp3SubmitRspPkt:
			c.mu.Lock()
			ch := c.pending[p.SeqId]
			c.mu.Unlock()
			if ch != nil {
				ch <- submitPartResponse{seqID: p.SeqId, msgID: p.MsgId, result: p.Result}
			}
		case *cmpp.Cmpp2DeliverReqPkt:
			_ = c.client.SendRspPkt(&cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
			c.handleDeliver(deliverPacketFromCMPP2(p))
		case *cmpp.Cmpp3DeliverReqPkt:
			_ = c.client.SendRspPkt(&cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
			c.handleDeliver(deliverPacketFromCMPP3(p))
		case *cmpp.CmppActiveTestReqPkt:
			_ = c.client.SendRspPkt(&cmpp.CmppActiveTestRspPkt{}, p.SeqId)
		}
	}
}

type deliverPacket struct {
	seqID            uint32
	msgID            uint64
	destID           string
	tpUdhi           uint8
	msgFmt           uint8
	srcTerminalID    string
	registerDelivery uint8
	msgContent       string
}

func deliverPacketFromCMPP2(pkt *cmpp.Cmpp2DeliverReqPkt) deliverPacket {
	return deliverPacket{
		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{
		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) {
	if pkt.registerDelivery == 1 {
		var receipt cmpp.CmppReceiptPkt
		if err := receipt.Unpack([]byte(pkt.msgContent)); err != nil {
			return
		}
		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(),
		}
		_ = postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/receipt", event)
		return
	}

	content, complete, err := c.decodeUplinkContent(pkt)
	if err != nil {
		return
	}
	if !complete {
		return
	}
	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,
		PhoneNumber: strings.TrimSpace(pkt.srcTerminalID),
		DestID:      strings.TrimSpace(pkt.destID),
		Content:     content,
		ReceivedAt:  time.Now().UTC(),
	}
	_ = postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/uplink", event)
}

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 (c *connection) close() {
	c.handleConnectionLoss(fmt.Errorf("connection closed"))
}

func (c *connection) handleConnectionLoss(err error) {
	c.mu.Lock()
	if c.closed && c.client == nil {
		c.mu.Unlock()
		return
	}
	c.closed = true
	pending := c.pending
	c.pending = make(map[uint32]chan submitPartResponse)
	if c.client != nil {
		c.client.Disconnect()
		c.client = nil
	}
	c.mu.Unlock()

	for _, ch := range pending {
		select {
		case ch <- submitPartResponse{err: err}:
		default:
		}
	}
	if c.pool != nil {
		status := "disconnected"
		if c.pool.countActiveConnections() > 0 {
			status = "connected"
		}
		_ = c.pool.reportState(context.Background(), status, err)
	}
}

func submitResult(cmd queue.SubmitCommand, sequenceID uint32, gatewayMessageID string, status string, code string, message string) queue.SubmitResult {
	if gatewayMessageID == "" {
		gatewayMessageID = fmt.Sprintf("GW-%s-%d", cmd.SubmitID, time.Now().UnixNano())
	}
	return queue.SubmitResult{
		Envelope: queue.Envelope{
			SchemaVersion: queue.SchemaVersion,
			MessageType:   queue.MessageTypeSubmitResult,
			TraceID:       cmd.TraceID,
			MessageID:     cmd.MessageID,
			ChannelID:     cmd.ChannelID,
			CreatedAt:     time.Now().UTC(),
		},
		SequenceID:       sequenceID,
		GatewayMessageID: gatewayMessageID,
		SubmitStatus:     status,
		ErrorCode:        code,
		ErrorMessage:     message,
		SubmittedAt:      time.Now().UTC(),
	}
}

func submitSegmentResult(part submitPart, sequenceID uint32, gatewayMessageID string, result queue.SubmitResult) queue.SubmitSegmentResult {
	if gatewayMessageID == "" {
		gatewayMessageID = result.GatewayMessageID
	}
	return queue.SubmitSegmentResult{
		SegmentTotal:     int(part.PkTotal),
		SegmentIndex:     int(part.PkNumber),
		SequenceID:       sequenceID,
		GatewayMessageID: gatewayMessageID,
		SubmitStatus:     result.SubmitStatus,
		ErrorCode:        result.ErrorCode,
		ErrorMessage:     result.ErrorMessage,
		SubmittedAt:      result.SubmittedAt,
	}
}

func validateSubmitCommand(cmd queue.SubmitCommand) error {
	if cmd.MessageType != queue.MessageTypeSubmitCommand {
		return fmt.Errorf("unsupported messageType %q", cmd.MessageType)
	}
	if cmd.MessageID == "" || cmd.ChannelID == "" || cmd.SubmitID == "" {
		return fmt.Errorf("messageId, channelId and submitId are required")
	}
	if cmd.Upstream.GatewayHost == "" || cmd.Upstream.GatewayPort <= 0 {
		return fmt.Errorf("upstream gatewayHost and gatewayPort are required")
	}
	if cmd.Upstream.Account == "" || cmd.Upstream.PasswordCipher == "" {
		return fmt.Errorf("upstream account and passwordCipher are required")
	}
	if len(cmd.PhoneNumber) == 0 || len(cmd.Content) == 0 {
		return fmt.Errorf("phoneNumber and content are required")
	}
	return nil
}

func validateConnectChannelCommand(command queue.ConnectChannelCommand) error {
	if command.MessageType != queue.MessageTypeConnectChannel {
		return fmt.Errorf("unsupported messageType %q", command.MessageType)
	}
	if command.ChannelID == "" || command.ConnectionID == "" {
		return fmt.Errorf("channelId and connectionId are required")
	}
	if command.Channel.GatewayHost == "" || command.Channel.GatewayPort <= 0 {
		return fmt.Errorf("gatewayHost and gatewayPort are required")
	}
	if command.Channel.Account == "" || command.Channel.PasswordCipher == "" {
		return fmt.Errorf("account and passwordCipher are required")
	}
	return nil
}

func defaultChannelConnectionID(channelID string) string {
	return fmt.Sprintf("%s:primary", channelID)
}

func normalizeUpstreamConfig(config queue.UpstreamConfig) queue.UpstreamConfig {
	if config.DesiredConnections <= 0 {
		config.DesiredConnections = 1
	}
	if config.WindowSize <= 0 {
		config.WindowSize = defaultWindowSize
	}
	return config
}

func isTemporaryReadTimeout(err error) bool {
	var netErr net.Error
	return errors.As(err, &netErr) && netErr.Timeout()
}

func postJSON(ctx context.Context, client *http.Client, apiBaseURL string, path string, payload any) error {
	if client == nil {
		client = &http.Client{Timeout: defaultHTTPTimeout}
	}
	body, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	base := strings.TrimRight(apiBaseURL, "/")
	if base == "" {
		base = "http://127.0.0.1:3000/api"
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, base+path, bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("api returned %s", resp.Status)
	}
	return nil
}

func encodeContent(format int, content string) (string, error) {
	switch format {
	case 8:
		return cmpputils.Utf8ToUcs2(content)
	case 15:
		return cmpputils.Utf8ToGB18030(content)
	default:
		return content, nil
	}
}

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 protocolVersion(version string) cmpp.Type {
	if strings.HasPrefix(version, "2") {
		return cmpp.V20
	}
	return cmpp.V30
}

func receiptStatus(stat string) string {
	switch strings.ToUpper(strings.TrimSpace(stat)) {
	case "DELIVRD":
		return "delivered"
	case "":
		return "unknown"
	default:
		return "undelivered"
	}
}

func defaultString(value string, fallback string) string {
	if value == "" {
		return fallback
	}
	return value
}

func defaultInt(value int, fallback int) int {
	if value == 0 {
		return fallback
	}
	return value
}