437 lines
16 KiB
Go
437 lines
16 KiB
Go
package control
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"time"
|
|
|
|
"cmpp-platform/gateway/internal/inbound"
|
|
"cmpp-platform/gateway/internal/queue"
|
|
"cmpp-platform/gateway/internal/ratelimit"
|
|
"cmpp-platform/gateway/internal/upstream"
|
|
)
|
|
|
|
type ConnectFunc func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error)
|
|
type DisconnectFunc func(context.Context, DisconnectChannelCommand) (ConnectionStateCallback, error)
|
|
type SubmitFunc func(context.Context, queue.SubmitCommand) (queue.SubmitResult, error)
|
|
|
|
type ConnectChannelCommand struct {
|
|
SchemaVersion string `json:"schemaVersion"`
|
|
MessageType string `json:"messageType"`
|
|
TraceID string `json:"traceId"`
|
|
ChannelID string `json:"channelId"`
|
|
ConnectionID string `json:"connectionId"`
|
|
Reason string `json:"reason"`
|
|
DesiredConnections int `json:"desiredConnections"`
|
|
Channel ChannelConfig `json:"channel"`
|
|
}
|
|
|
|
type ChannelConfig struct {
|
|
Code string `json:"code"`
|
|
Name string `json:"name"`
|
|
GatewayHost string `json:"gatewayHost"`
|
|
GatewayPort int `json:"gatewayPort"`
|
|
Account string `json:"account"`
|
|
PasswordCipher string `json:"passwordCipher"`
|
|
SrcID string `json:"srcId"`
|
|
CMPPVersion string `json:"cmppVersion"`
|
|
RateLimitPerSecond int `json:"rateLimitPerSecond"`
|
|
WindowSize int `json:"windowSize,omitempty"`
|
|
HeartbeatIntervalSeconds int `json:"heartbeatIntervalSeconds,omitempty"`
|
|
HeartbeatMissThreshold int `json:"heartbeatMissThreshold,omitempty"`
|
|
ConnectionWarmupSeconds int `json:"connectionWarmupSeconds,omitempty"`
|
|
ConnectionDrainSeconds int `json:"connectionDrainTimeoutSeconds,omitempty"`
|
|
SubmitTimeoutSeconds int `json:"submitResponseTimeoutSeconds,omitempty"`
|
|
FailureCooldownSeconds int `json:"connectionFailureCooldownSeconds,omitempty"`
|
|
}
|
|
|
|
type DisconnectChannelCommand struct {
|
|
SchemaVersion string `json:"schemaVersion"`
|
|
MessageType string `json:"messageType"`
|
|
TraceID string `json:"traceId"`
|
|
ChannelID string `json:"channelId"`
|
|
ConnectionID string `json:"connectionId"`
|
|
CreatedAt string `json:"createdAt"`
|
|
Reason string `json:"reason"`
|
|
}
|
|
|
|
type ConnectionStateCallback 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"`
|
|
}
|
|
|
|
type Server struct {
|
|
APIBaseURL string
|
|
HTTPClient *http.Client
|
|
Connect ConnectFunc
|
|
Disconnect DisconnectFunc
|
|
Submit SubmitFunc
|
|
Upstream *upstream.Manager
|
|
Limiter ratelimit.Limiter
|
|
RecoveryCandidates func(context.Context) ([]inbound.DownstreamPresence, error)
|
|
RecoveryStatuses func(context.Context) ([]inbound.DownstreamRecoveryStatus, error)
|
|
}
|
|
|
|
type DownstreamRecoveryOverview struct {
|
|
Candidates []inbound.DownstreamPresence `json:"candidates"`
|
|
Statuses []inbound.DownstreamRecoveryStatus `json:"statuses"`
|
|
}
|
|
|
|
type DisconnectDownstreamAccountCommand struct {
|
|
Account string `json:"account"`
|
|
Reason string `json:"reason"`
|
|
}
|
|
|
|
func Register(mux *http.ServeMux, server Server) {
|
|
if server.HTTPClient == nil {
|
|
server.HTTPClient = &http.Client{Timeout: 10 * time.Second}
|
|
}
|
|
if server.Upstream == nil {
|
|
server.Upstream = &upstream.Manager{APIBaseURL: server.APIBaseURL, HTTPClient: server.HTTPClient}
|
|
}
|
|
if server.Connect == nil {
|
|
server.Connect = server.connectChannel
|
|
}
|
|
if server.Disconnect == nil {
|
|
server.Disconnect = server.disconnectChannel
|
|
}
|
|
if server.Submit == nil {
|
|
server.Submit = server.Upstream.Submit
|
|
}
|
|
mux.HandleFunc("/connections/connect", server.handleConnectChannel)
|
|
mux.HandleFunc("/connections/disconnect", server.handleDisconnectChannel)
|
|
mux.HandleFunc("/upstream/submit", server.handleUpstreamSubmit)
|
|
mux.HandleFunc("/downstream/receipt", server.handleDownstreamReceipt)
|
|
mux.HandleFunc("/downstream/uplink", server.handleDownstreamUplink)
|
|
mux.HandleFunc("/downstream/recovery-candidates", server.handleDownstreamRecoveryCandidates)
|
|
mux.HandleFunc("/downstream/recovery-statuses", server.handleDownstreamRecoveryStatuses)
|
|
mux.HandleFunc("/downstream/recovery-overview", server.handleDownstreamRecoveryOverview)
|
|
mux.HandleFunc("/downstream/connections/disconnect", server.handleDisconnectDownstreamAccount)
|
|
}
|
|
|
|
func (s Server) handleDisconnectDownstreamAccount(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
var command DisconnectDownstreamAccountCommand
|
|
if err := json.NewDecoder(r.Body).Decode(&command); err != nil {
|
|
http.Error(w, fmt.Sprintf("invalid downstream disconnect command: %v", err), http.StatusBadRequest)
|
|
return
|
|
}
|
|
if command.Account == "" {
|
|
http.Error(w, "account is required", http.StatusBadRequest)
|
|
return
|
|
}
|
|
disconnected := inbound.DisconnectAccount(command.Account)
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(map[string]any{
|
|
"account": command.Account,
|
|
"disconnected": disconnected,
|
|
"reason": command.Reason,
|
|
})
|
|
}
|
|
|
|
func (s Server) handleConnectChannel(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
|
|
var command ConnectChannelCommand
|
|
if err := json.NewDecoder(r.Body).Decode(&command); err != nil {
|
|
http.Error(w, fmt.Sprintf("invalid connect command: %v", err), http.StatusBadRequest)
|
|
return
|
|
}
|
|
if err := validateConnectChannelCommand(command); err != nil {
|
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
if configurer, ok := s.Limiter.(ratelimit.Configurer); ok {
|
|
if err := configurer.Configure(r.Context(), command.ChannelID, command.Channel.RateLimitPerSecond); err != nil {
|
|
http.Error(w, fmt.Sprintf("failed to configure gateway channel rate limit: %v", err), http.StatusServiceUnavailable)
|
|
return
|
|
}
|
|
}
|
|
|
|
status, err := s.Connect(r.Context(), command)
|
|
if err != nil {
|
|
http.Error(w, fmt.Sprintf("failed to establish upstream pool: %v", err), http.StatusBadGateway)
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(status)
|
|
}
|
|
|
|
func (s Server) handleDisconnectChannel(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
var command DisconnectChannelCommand
|
|
if err := json.NewDecoder(r.Body).Decode(&command); err != nil {
|
|
http.Error(w, fmt.Sprintf("invalid disconnect command: %v", err), http.StatusBadRequest)
|
|
return
|
|
}
|
|
if command.MessageType != string(queue.MessageTypeDisconnectChannel) || command.ChannelID == "" || command.ConnectionID == "" {
|
|
http.Error(w, "messageType DisconnectChannel, channelId and connectionId are required", http.StatusBadRequest)
|
|
return
|
|
}
|
|
status, err := s.Disconnect(r.Context(), command)
|
|
if err != nil {
|
|
http.Error(w, fmt.Sprintf("failed to disconnect upstream pool: %v", err), http.StatusBadGateway)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(status)
|
|
}
|
|
|
|
func (s Server) handleUpstreamSubmit(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
var command queue.SubmitCommand
|
|
if err := json.NewDecoder(r.Body).Decode(&command); err != nil {
|
|
http.Error(w, fmt.Sprintf("invalid submit command: %v", err), http.StatusBadRequest)
|
|
return
|
|
}
|
|
if s.Limiter != nil {
|
|
if _, err := s.Limiter.Wait(r.Context(), command.ChannelID, command.Route.RateLimitPerSecond); err != nil {
|
|
http.Error(w, fmt.Sprintf("gateway channel rate limit unavailable: %v", err), http.StatusServiceUnavailable)
|
|
return
|
|
}
|
|
}
|
|
result, err := s.Submit(r.Context(), command)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadGateway)
|
|
_ = json.NewEncoder(w).Encode(result)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(result)
|
|
}
|
|
|
|
func (s Server) handleDownstreamReceipt(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
var event inbound.DownstreamReceipt
|
|
if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
|
|
http.Error(w, fmt.Sprintf("invalid downstream receipt: %v", err), http.StatusBadRequest)
|
|
return
|
|
}
|
|
result, err := inbound.PushReceiptWithResult(event)
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(result)
|
|
}
|
|
|
|
func (s Server) handleDownstreamUplink(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
var event inbound.DownstreamUplink
|
|
if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
|
|
http.Error(w, fmt.Sprintf("invalid downstream uplink: %v", err), http.StatusBadRequest)
|
|
return
|
|
}
|
|
result, err := inbound.PushUplinkWithResult(event)
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(result)
|
|
}
|
|
|
|
func (s Server) handleDownstreamRecoveryCandidates(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
if s.RecoveryCandidates == nil {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode([]inbound.DownstreamPresence{})
|
|
return
|
|
}
|
|
candidates, err := s.RecoveryCandidates(r.Context())
|
|
if err != nil {
|
|
http.Error(w, fmt.Sprintf("failed to load recovery candidates: %v", err), http.StatusBadGateway)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(candidates)
|
|
}
|
|
|
|
func (s Server) handleDownstreamRecoveryStatuses(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
if s.RecoveryStatuses == nil {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode([]inbound.DownstreamRecoveryStatus{})
|
|
return
|
|
}
|
|
statuses, err := s.RecoveryStatuses(r.Context())
|
|
if err != nil {
|
|
http.Error(w, fmt.Sprintf("failed to load recovery statuses: %v", err), http.StatusBadGateway)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(statuses)
|
|
}
|
|
|
|
func (s Server) handleDownstreamRecoveryOverview(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
candidates := []inbound.DownstreamPresence{}
|
|
if s.RecoveryCandidates != nil {
|
|
result, err := s.RecoveryCandidates(r.Context())
|
|
if err != nil {
|
|
http.Error(w, fmt.Sprintf("failed to load recovery candidates: %v", err), http.StatusBadGateway)
|
|
return
|
|
}
|
|
candidates = result
|
|
}
|
|
statuses := []inbound.DownstreamRecoveryStatus{}
|
|
if s.RecoveryStatuses != nil {
|
|
result, err := s.RecoveryStatuses(r.Context())
|
|
if err != nil {
|
|
http.Error(w, fmt.Sprintf("failed to load recovery statuses: %v", err), http.StatusBadGateway)
|
|
return
|
|
}
|
|
statuses = result
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(DownstreamRecoveryOverview{
|
|
Candidates: candidates,
|
|
Statuses: statuses,
|
|
})
|
|
}
|
|
|
|
func (s Server) connectChannel(ctx context.Context, command ConnectChannelCommand) (ConnectionStateCallback, error) {
|
|
state, err := s.Upstream.ConnectChannel(ctx, queue.ConnectChannelCommand{
|
|
SchemaVersion: command.SchemaVersion,
|
|
MessageType: queue.MessageTypeConnectChannel,
|
|
TraceID: command.TraceID,
|
|
ChannelID: command.ChannelID,
|
|
ConnectionID: command.ConnectionID,
|
|
CreatedAt: time.Now().UTC(),
|
|
Reason: command.Reason,
|
|
DesiredConnections: command.DesiredConnections,
|
|
Channel: queue.ConnectChannelConfig{
|
|
Code: command.Channel.Code,
|
|
Name: command.Channel.Name,
|
|
GatewayHost: command.Channel.GatewayHost,
|
|
GatewayPort: command.Channel.GatewayPort,
|
|
Account: command.Channel.Account,
|
|
PasswordCipher: command.Channel.PasswordCipher,
|
|
SrcID: command.Channel.SrcID,
|
|
CMPPVersion: command.Channel.CMPPVersion,
|
|
RateLimitPerSecond: command.Channel.RateLimitPerSecond,
|
|
WindowSize: command.Channel.WindowSize,
|
|
HeartbeatIntervalSeconds: command.Channel.HeartbeatIntervalSeconds,
|
|
HeartbeatMissThreshold: command.Channel.HeartbeatMissThreshold,
|
|
ConnectionWarmupSeconds: command.Channel.ConnectionWarmupSeconds,
|
|
ConnectionDrainSeconds: command.Channel.ConnectionDrainSeconds,
|
|
SubmitTimeoutSeconds: command.Channel.SubmitTimeoutSeconds,
|
|
FailureCooldownSeconds: command.Channel.FailureCooldownSeconds,
|
|
},
|
|
})
|
|
if err != nil {
|
|
return ConnectionStateCallback{}, err
|
|
}
|
|
return ConnectionStateCallback{
|
|
ChannelID: state.ChannelID,
|
|
ConnectionID: state.ConnectionID,
|
|
Status: state.Status,
|
|
DesiredConnections: state.DesiredConnections,
|
|
CurrentConnections: state.CurrentConnections,
|
|
LastConnectedAt: state.LastConnectedAt,
|
|
LastDisconnectedAt: state.LastDisconnectedAt,
|
|
LastHeartbeatAt: state.LastHeartbeatAt,
|
|
ReconnectCount: state.ReconnectCount,
|
|
LastReconnectAttemptAt: state.LastReconnectAttemptAt,
|
|
NextReconnectAt: state.NextReconnectAt,
|
|
LastErrorCategory: state.LastErrorCategory,
|
|
LastError: state.LastError,
|
|
}, nil
|
|
}
|
|
|
|
func (s Server) disconnectChannel(ctx context.Context, command DisconnectChannelCommand) (ConnectionStateCallback, error) {
|
|
state, err := s.Upstream.DisconnectChannel(ctx, queue.DisconnectChannelCommand{
|
|
SchemaVersion: command.SchemaVersion,
|
|
MessageType: queue.MessageTypeDisconnectChannel,
|
|
TraceID: command.TraceID,
|
|
ChannelID: command.ChannelID,
|
|
ConnectionID: command.ConnectionID,
|
|
CreatedAt: time.Now().UTC(),
|
|
Reason: command.Reason,
|
|
})
|
|
if err != nil {
|
|
return ConnectionStateCallback{}, err
|
|
}
|
|
return ConnectionStateCallback{
|
|
ChannelID: state.ChannelID,
|
|
ConnectionID: state.ConnectionID,
|
|
Status: state.Status,
|
|
DesiredConnections: state.DesiredConnections,
|
|
CurrentConnections: state.CurrentConnections,
|
|
LastDisconnectedAt: state.LastDisconnectedAt,
|
|
ReconnectCount: state.ReconnectCount,
|
|
LastError: state.LastError,
|
|
}, nil
|
|
}
|
|
|
|
func validateConnectChannelCommand(command ConnectChannelCommand) error {
|
|
if command.MessageType != "ConnectChannel" {
|
|
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")
|
|
}
|
|
if command.DesiredConnections < 1 || command.DesiredConnections > 8 {
|
|
return fmt.Errorf("desiredConnections must be between 1 and 8")
|
|
}
|
|
if command.Channel.WindowSize != 0 && (command.Channel.WindowSize < 1 || command.Channel.WindowSize > 64) {
|
|
return fmt.Errorf("windowSize must be between 1 and 64")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func desiredConnections(value int) int {
|
|
if value > 0 {
|
|
return value
|
|
}
|
|
return 1
|
|
}
|