lislgosms/gateway/internal/upstream/manager.go

package upstream

import (
	"cmpp-platform/gateway/internal/queue"
	"context"
	"fmt"
	"net/http"
	"sync"
	"time"
)

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
	SubmitSegmentPublisher SubmitSegmentPublisher

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

// SubmitSegmentPublisher persists each supplier response before the next long-message
// segment is sent. The boundary is intentionally storage-only: HTTP callbacks belong to
// the result Outbox worker and must not consume a supplier Submit window slot.
type SubmitSegmentPublisher interface {
	PublishSubmitSegment(context.Context, queue.SubmitCommand, queue.SubmitSegmentResult) error
}

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) 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) ensureDefaultsLocked() {
	if m.HTTPClient == nil {
		m.HTTPClient = &http.Client{Timeout: defaultHTTPTimeout}
	}
	if m.conns == nil {
		m.conns = make(map[string]*connectionPool)
	}
}

// ConnectionCounts returns bounded platform totals without channel identifiers to prevent time-series cardinality growth.
func (m *Manager) ConnectionCounts() (desired int, connected int) {
	m.mu.Lock()
	pools := make([]*connectionPool, 0, len(m.conns))
	for _, pool := range m.conns {
		pools = append(pools, pool)
	}
	m.mu.Unlock()
	for _, pool := range pools {
		desired += max(pool.config.DesiredConnections, 1)
		connected += pool.countActiveConnections()
	}
	return desired, connected
}

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{}),
	}
}

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
}