lislgosms/gateway/internal/upstream/connection.go

package upstream

import (
	"cmpp-platform/gateway/internal/protocollog"
	"cmpp-platform/gateway/internal/queue"
	"context"
	"fmt"
	cmpp "github.com/bigwhite/gocmpp"
	"log"
	"net/http"
	"strings"
	"sync"
	"time"
)

// The reader and heartbeat goroutines share one connection lifecycle. Closing
// the connection must wake pending submitters before scheduling pool recovery.

// Caller holds c.mu across sequence selection, wire write and pending registration.
func (c *connection) sequenceAvailable(sequence uint32) bool {
	_, submit := c.pending[sequence]
	_, heartbeat := c.heartbeatPending[sequence]
	return !submit && !heartbeat
}

type connection struct {
	channelID            string
	config               queue.UpstreamConfig
	index                int
	pool                 *connectionPool
	apiBaseURL           string
	httpClient           *http.Client
	protocolLogPublisher interface {
		Publish(context.Context, protocollog.Event) error
	}
	gatewayInstanceID string
	eventPublisher    interface {
		PublishReceipt(context.Context, queue.ReceiptEvent) error
		PublishUplink(context.Context, queue.UplinkEvent) error
	}

	mu                  sync.Mutex
	sendMu              sync.Mutex
	client              *cmpp.Client
	window              chan struct{}
	windowLimit         int
	draining            bool
	retired             bool
	lastSubmitRTT       time.Duration
	consecutiveFailures int
	cooldownUntil       time.Time
	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) 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:
			deliver := deliverPacketFromCMPP2(p)
			if deliver.registerDelivery == 1 {
				if err := c.handleDeliver(deliver); err != nil {
					log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_receipt status=intake_failed channel_id=%s sequence_id=%d error=%q", c.channelID, deliver.seqID, err)
					c.handleConnectionLoss(fmt.Errorf("persist upstream receipt before DELIVER_RESP: %w", err))
					return
				}
				responseErr := c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
				c.emitDeliverResponse(deliver, responseErr)
			} else {
				responseErr := c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
				c.emitDeliverResponse(deliver, responseErr)
				_ = c.handleDeliver(deliver)
			}
		case *cmpp.Cmpp3DeliverReqPkt:
			deliver := deliverPacketFromCMPP3(p)
			if deliver.registerDelivery == 1 {
				if err := c.handleDeliver(deliver); err != nil {
					log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_receipt status=intake_failed channel_id=%s sequence_id=%d error=%q", c.channelID, deliver.seqID, err)
					c.handleConnectionLoss(fmt.Errorf("persist upstream receipt before DELIVER_RESP: %w", err))
					return
				}
				responseErr := c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
				c.emitDeliverResponse(deliver, responseErr)
			} else {
				responseErr := c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
				c.emitDeliverResponse(deliver, responseErr)
				_ = c.handleDeliver(deliver)
			}
		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) identity() string {
	if c.pool != nil && strings.TrimSpace(c.pool.connectionID) != "" {
		return fmt.Sprintf("%s-%d", c.pool.connectionID, c.index)
	}
	return fmt.Sprintf("%s-%d", c.channelID, c.index)
}

func (c *connection) retire() {
	c.mu.Lock()
	c.draining = true
	c.retired = true
	c.mu.Unlock()
	c.handleConnectionLoss(fmt.Errorf("connection retired after drain"))
}

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 && !c.retired {
		status := "disconnected"
		if c.pool.countActiveConnections() > 0 {
			status = "reconnecting"
		}
		c.pool.scheduleReconnect(err)
		_ = c.pool.reportState(context.Background(), status, err)
	}
}