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
	defaultHeartbeatInterval      = 30 * time.Second
	defaultHeartbeatMissThreshold = 3
	defaultReconnectInitialDelay  = 5 * time.Second
	defaultReconnectMaximumDelay  = 5 * time.Minute
	defaultAuthReconnectDelay     = 5 * time.Minute
)

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"`
	LastReconnectAttemptAt string `json:"lastReconnectAttemptAt,omitempty"`
	NextReconnectAt        string `json:"nextReconnectAt,omitempty"`
	LastErrorCategory      string `json:"lastErrorCategory,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:               command.Channel.WindowSize,
		HeartbeatIntervalSeconds: command.Channel.HeartbeatIntervalSeconds,
		HeartbeatMissThreshold:   command.Channel.HeartbeatMissThreshold,
	})
	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()

	pool.startSupervisor()
	if err := pool.ensureConnected(); err != nil {
		if pool.stopped() {
			return pool.snapshotState("disconnected", nil), nil
		}
		pool.scheduleReconnect(err)
		_ = pool.reportState(ctx, "failed", err)
		return pool.snapshotState("failed", err), nil
	}
	if pool.stopped() {
		return pool.snapshotState("disconnected", nil), nil
	}
	pool.resetReconnectState()
	return pool.snapshotState("connected", nil), nil
}

func (m *Manager) DisconnectChannel(ctx context.Context, command queue.DisconnectChannelCommand) (ConnectionState, error) {
	if command.ChannelID == "" || command.ConnectionID == "" {
		return ConnectionState{}, fmt.Errorf("channelId and connectionId are required")
	}
	m.mu.Lock()
	m.ensureDefaultsLocked()
	pool := m.conns[command.ChannelID]
	delete(m.conns, command.ChannelID)
	m.mu.Unlock()
	if pool != nil {
		pool.close()
		state := pool.snapshotState("disconnected", nil)
		_ = pool.reportState(ctx, "disconnected", nil)
		return state, nil
	}
	return ConnectionState{
		ChannelID:          command.ChannelID,
		ConnectionID:       command.ConnectionID,
		Status:             "disconnected",
		DesiredConnections: 0,
		CurrentConnections: 0,
		LastDisconnectedAt: time.Now().UTC().Format(time.RFC3339Nano),
	}, 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
	}
	pool.startSupervisor()
	if err := pool.ensureConnected(); err != nil {
		pool.scheduleReconnect(err)
		return nil, err
	}
	pool.resetReconnectState()
	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)
		},
		reconnectSignal: make(chan struct{}, 1),
		stopCh:          make(chan struct{}),
	}
}

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

	mu                     sync.Mutex
	connectMu              sync.Mutex
	conns                  []*connection
	next                   int
	reconnectSignal        chan struct{}
	stopCh                 chan struct{}
	stopOnce               sync.Once
	supervisorOnce         sync.Once
	reconnectCount         int
	lastReconnectAttemptAt time.Time
	nextReconnectAt        time.Time
	lastErrorCategory      string
}

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

func (p *connectionPool) ensureConnected() error {
	p.connectMu.Lock()
	defer p.connectMu.Unlock()
	desired := p.config.DesiredConnections
	if desired <= 0 {
		desired = 1
	}
	connectedAny := false
	for {
		p.mu.Lock()
		active := p.conns[:0]
		for _, existing := range p.conns {
			existing.mu.Lock()
			usable := existing.client != nil && !existing.closed
			existing.mu.Unlock()
			if usable {
				active = append(active, existing)
			}
		}
		p.conns = active
		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),
			heartbeatPending: make(map[uint32]time.Time),
		}
		p.mu.Unlock()

		connected, err := conn.ensureConnected()
		if err != nil {
			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) startSupervisor() {
	p.supervisorOnce.Do(func() {
		go p.superviseReconnects()
	})
}

func (p *connectionPool) stopped() bool {
	select {
	case <-p.stopCh:
		return true
	default:
		return false
	}
}

func (p *connectionPool) signalReconnect() {
	select {
	case <-p.stopCh:
		return
	default:
	}
	select {
	case p.reconnectSignal <- struct{}{}:
	default:
	}
}

func (p *connectionPool) scheduleReconnect(stateErr error) {
	select {
	case <-p.stopCh:
		return
	default:
	}
	p.mu.Lock()
	now := time.Now().UTC()
	if !p.nextReconnectAt.IsZero() && p.nextReconnectAt.After(now) {
		p.mu.Unlock()
		return
	}
	p.reconnectCount++
	p.lastReconnectAttemptAt = now
	p.lastErrorCategory = connectionErrorCategory(stateErr)
	delay := reconnectDelay(p.reconnectCount, p.lastErrorCategory)
	p.nextReconnectAt = p.lastReconnectAttemptAt.Add(delay)
	p.mu.Unlock()
	p.signalReconnect()
}

func (p *connectionPool) resetReconnectState() {
	p.mu.Lock()
	p.reconnectCount = 0
	p.lastReconnectAttemptAt = time.Time{}
	p.nextReconnectAt = time.Time{}
	p.lastErrorCategory = ""
	p.mu.Unlock()
	p.signalReconnect()
}

func (p *connectionPool) superviseReconnects() {
	for {
		select {
		case <-p.stopCh:
			return
		case <-p.reconnectSignal:
		}
		for {
			p.mu.Lock()
			next := p.nextReconnectAt
			p.mu.Unlock()
			if next.IsZero() {
				break
			}
			timer := time.NewTimer(time.Until(next))
			select {
			case <-p.stopCh:
				if !timer.Stop() {
					<-timer.C
				}
				return
			case <-p.reconnectSignal:
				if !timer.Stop() {
					<-timer.C
				}
				continue
			case <-timer.C:
			}
			_ = p.reportState(context.Background(), "reconnecting", nil)
			p.mu.Lock()
			p.lastReconnectAttemptAt = time.Now().UTC()
			p.mu.Unlock()
			if err := p.ensureConnected(); err != nil {
				select {
				case <-p.stopCh:
					return
				default:
				}
				p.scheduleReconnect(err)
				_ = p.reportState(context.Background(), "failed", err)
				continue
			}
			select {
			case <-p.stopCh:
				return
			default:
			}
			p.resetReconnectState()
			_ = p.reportState(context.Background(), "connected", nil)
			break
		}
	}
}

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.connectMu.Lock()
	defer p.connectMu.Unlock()
	p.stopOnce.Do(func() {
		close(p.stopCh)
	})
	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(),
	}
	p.mu.Lock()
	state.ReconnectCount = p.reconnectCount
	state.LastErrorCategory = p.lastErrorCategory
	if !p.lastReconnectAttemptAt.IsZero() {
		state.LastReconnectAttemptAt = p.lastReconnectAttemptAt.Format(time.RFC3339Nano)
	}
	if !p.nextReconnectAt.IsZero() {
		state.NextReconnectAt = p.nextReconnectAt.Format(time.RFC3339Nano)
	}
	p.mu.Unlock()
	if state.DesiredConnections <= 0 {
		state.DesiredConnections = 1
	}
	switch status {
	case "connected":
		state.LastConnectedAt = now
		state.LastHeartbeatAt = now
	case "heartbeat":
		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
	heartbeatCancel  context.CancelFunc
	heartbeatPending map[uint32]time.Time
}

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
	c.heartbeatPending = make(map[uint32]time.Time)
	heartbeatCtx, cancelHeartbeat := context.WithCancel(context.Background())
	c.heartbeatCancel = cancelHeartbeat
	go c.readLoop()
	go c.heartbeatLoop(heartbeatCtx)
	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.mu.Lock()
	client := c.client
	closed := c.closed
	c.mu.Unlock()
	if closed || client == nil {
		err := fmt.Errorf("supplier connection is not available")
		result := submitResult(cmd, 0, "", "timeout", "CONNECTION_LOST", err.Error())
		return 0, "", result, err
	}
	c.sendMu.Lock()
	seq, err := 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 {
		c.mu.Lock()
		client := c.client
		closed := c.closed
		c.mu.Unlock()
		if closed || client == nil {
			return
		}
		pkt, err := 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)
			return
		}
		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.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
			c.handleDeliver(deliverPacketFromCMPP2(p))
		case *cmpp.Cmpp3DeliverReqPkt:
			_ = c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
			c.handleDeliver(deliverPacketFromCMPP3(p))
		case *cmpp.CmppActiveTestReqPkt:
			_ = c.sendResponse(client, &cmpp.CmppActiveTestRspPkt{}, p.SeqId)
			_ = c.pool.reportState(context.Background(), "heartbeat", nil)
		case *cmpp.CmppActiveTestRspPkt:
			c.handleHeartbeatResponse(p.SeqId)
			_ = c.pool.reportState(context.Background(), "heartbeat", nil)
		}
	}
}

func (c *connection) sendResponse(client *cmpp.Client, packet cmpp.Packer, sequenceID uint32) error {
	c.sendMu.Lock()
	defer c.sendMu.Unlock()
	return client.SendRspPkt(packet, sequenceID)
}

func (c *connection) heartbeatLoop(ctx context.Context) {
	interval := time.Duration(c.config.HeartbeatIntervalSeconds) * time.Second
	if interval <= 0 {
		interval = defaultHeartbeatInterval
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if !c.sendHeartbeat() {
				return
			}
		}
	}
}

func (c *connection) sendHeartbeat() bool {
	threshold := c.config.HeartbeatMissThreshold
	if threshold <= 0 {
		threshold = defaultHeartbeatMissThreshold
	}
	c.mu.Lock()
	if c.closed {
		c.mu.Unlock()
		return false
	}
	if len(c.heartbeatPending) >= threshold {
		c.mu.Unlock()
		c.handleConnectionLoss(fmt.Errorf("heartbeat timeout after %d unanswered ACTIVE_TEST requests", threshold))
		return false
	}
	if c.client == nil {
		c.mu.Unlock()
		return false
	}
	c.sendMu.Lock()
	seq, err := c.client.SendReqPkt(&cmpp.CmppActiveTestReqPkt{})
	c.sendMu.Unlock()
	if err != nil {
		c.mu.Unlock()
		c.handleConnectionLoss(fmt.Errorf("send ACTIVE_TEST: %w", err))
		return false
	}
	if !c.closed {
		c.heartbeatPending[seq] = time.Now().UTC()
	}
	c.mu.Unlock()
	return true
}

func (c *connection) handleHeartbeatResponse(sequenceID uint32) {
	c.mu.Lock()
	delete(c.heartbeatPending, sequenceID)
	c.mu.Unlock()
}

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
	if c.heartbeatCancel != nil {
		c.heartbeatCancel()
		c.heartbeatCancel = nil
	}
	pending := c.pending
	c.pending = make(map[uint32]chan submitPartResponse)
	c.heartbeatPending = make(map[uint32]time.Time)
	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 = "reconnecting"
		}
		c.pool.scheduleReconnect(err)
		_ = 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
	}
	if config.HeartbeatIntervalSeconds <= 0 {
		config.HeartbeatIntervalSeconds = int(defaultHeartbeatInterval / time.Second)
	}
	if config.HeartbeatMissThreshold <= 0 {
		config.HeartbeatMissThreshold = defaultHeartbeatMissThreshold
	}
	return config
}

func connectionErrorCategory(err error) string {
	if err == nil {
		return ""
	}
	message := strings.ToLower(err.Error())
	switch {
	case strings.Contains(message, "auth"), strings.Contains(message, "password"), strings.Contains(message, "credential"):
		return "authentication"
	case strings.Contains(message, "heartbeat"):
		return "heartbeat_timeout"
	case strings.Contains(message, "timeout"):
		return "timeout"
	default:
		return "network"
	}
}

func reconnectDelay(attempt int, category string) time.Duration {
	if category == "authentication" {
		return defaultAuthReconnectDelay
	}
	if attempt < 1 {
		attempt = 1
	}
	delays := []time.Duration{
		defaultReconnectInitialDelay,
		15 * time.Second,
		30 * time.Second,
		time.Minute,
		2 * time.Minute,
		defaultReconnectMaximumDelay,
	}
	delay := delays[min(attempt-1, len(delays)-1)]
	// Deterministic ±10% jitter prevents a large set of channels from retrying together.
	offsetPercent := (attempt*37)%21 - 10
	delay += time.Duration(int64(delay) * int64(offsetPercent) / 100)
	if delay < time.Second {
		return time.Second
	}
	return delay
}

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
}