334 lines
11 KiB
Go
334 lines
11 KiB
Go
package control
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
|
|
"cmpp-platform/gateway/internal/inbound"
|
|
"cmpp-platform/gateway/internal/queue"
|
|
"cmpp-platform/gateway/internal/upstream"
|
|
|
|
cmpp "github.com/bigwhite/gocmpp"
|
|
)
|
|
|
|
const defaultConnectTimeout = 5 * time.Second
|
|
|
|
type DialFunc func(context.Context, ConnectChannelCommand) 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"`
|
|
}
|
|
|
|
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"`
|
|
LastError string `json:"lastError,omitempty"`
|
|
}
|
|
|
|
type Server struct {
|
|
APIBaseURL string
|
|
HTTPClient *http.Client
|
|
Dial DialFunc
|
|
Upstream *upstream.Manager
|
|
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"`
|
|
}
|
|
|
|
func Register(mux *http.ServeMux, server Server) {
|
|
if server.HTTPClient == nil {
|
|
server.HTTPClient = &http.Client{Timeout: 10 * time.Second}
|
|
}
|
|
if server.Dial == nil {
|
|
server.Dial = DialCMPP
|
|
}
|
|
if server.Upstream == nil {
|
|
server.Upstream = &upstream.Manager{APIBaseURL: server.APIBaseURL, HTTPClient: server.HTTPClient}
|
|
}
|
|
mux.HandleFunc("/connections/connect", server.handleConnectChannel)
|
|
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)
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
status := ConnectionStateCallback{
|
|
ChannelID: command.ChannelID,
|
|
ConnectionID: command.ConnectionID,
|
|
DesiredConnections: desiredConnections(command.DesiredConnections),
|
|
}
|
|
if err := s.Dial(r.Context(), command); err != nil {
|
|
status.Status = "failed"
|
|
status.CurrentConnections = 0
|
|
status.LastDisconnectedAt = time.Now().UTC().Format(time.RFC3339Nano)
|
|
status.LastError = err.Error()
|
|
} else {
|
|
now := time.Now().UTC().Format(time.RFC3339Nano)
|
|
status.Status = "connected"
|
|
status.CurrentConnections = status.DesiredConnections
|
|
status.LastConnectedAt = now
|
|
status.LastHeartbeatAt = now
|
|
}
|
|
|
|
if err := s.postConnectionState(r.Context(), status); err != nil {
|
|
http.Error(w, fmt.Sprintf("failed to callback api: %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
|
|
}
|
|
result, err := s.Upstream.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
|
|
}
|
|
delivered, err := inbound.PushReceipt(event)
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(map[string]any{"delivered": delivered})
|
|
}
|
|
|
|
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
|
|
}
|
|
delivered, err := inbound.PushUplink(event)
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(map[string]any{"delivered": delivered})
|
|
}
|
|
|
|
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 DialCMPP(ctx context.Context, command ConnectChannelCommand) error {
|
|
ctx, cancel := context.WithTimeout(ctx, defaultConnectTimeout)
|
|
defer cancel()
|
|
|
|
version := cmpp.V30
|
|
if strings.HasPrefix(command.Channel.CMPPVersion, "2") {
|
|
version = cmpp.V20
|
|
}
|
|
|
|
client := cmpp.NewClient(version)
|
|
defer client.Disconnect()
|
|
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
addr := fmt.Sprintf("%s:%d", command.Channel.GatewayHost, command.Channel.GatewayPort)
|
|
done <- client.Connect(addr, command.Channel.Account, command.Channel.PasswordCipher, defaultConnectTimeout)
|
|
}()
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case err := <-done:
|
|
return err
|
|
}
|
|
}
|
|
|
|
func (s Server) postConnectionState(ctx context.Context, state ConnectionStateCallback) error {
|
|
apiBaseURL := strings.TrimRight(s.APIBaseURL, "/")
|
|
if apiBaseURL == "" {
|
|
apiBaseURL = "http://127.0.0.1:3000/api"
|
|
}
|
|
payload, err := json.Marshal(state)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiBaseURL+"/admin/gateway/connections", bytes.NewReader(payload))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req.Header.Set("Content-Type", "application/json")
|
|
resp, err := s.HTTPClient.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 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")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func desiredConnections(value int) int {
|
|
if value > 0 {
|
|
return value
|
|
}
|
|
return 1
|
|
}
|