diff --git a/apps/carrier-connector/internal/handler/handlers.go b/apps/carrier-connector/internal/handlers/handlers.go similarity index 100% rename from apps/carrier-connector/internal/handler/handlers.go rename to apps/carrier-connector/internal/handlers/handlers.go diff --git a/apps/carrier-connector/internal/handler/profile_handlers.go b/apps/carrier-connector/internal/handlers/profile_handlers.go similarity index 100% rename from apps/carrier-connector/internal/handler/profile_handlers.go rename to apps/carrier-connector/internal/handlers/profile_handlers.go diff --git a/apps/carrier-connector/internal/handlers/smdp_handler.go b/apps/carrier-connector/internal/handlers/smdp_handler.go new file mode 100644 index 0000000..7c5d62e --- /dev/null +++ b/apps/carrier-connector/internal/handlers/smdp_handler.go @@ -0,0 +1,134 @@ +package handlers + +import ( + "context" + "net/http" + "time" + + "github.com/gin-gonic/gin" + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository" + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/smdp" +) + +// SMDPHandler handles SM-DP+ multi-carrier operations +type SMDPHandler struct { + manager *smdp.SMDPManager +} + +// NewSMDPHandler creates a new SM-DP+ handler +func NewSMDPHandler(repo *repository.PostgresProfileStore) *SMDPHandler { + config := smdp.DefaultManagerConfig() + manager := smdp.NewSMDPManager(repo, config) + + // Start health checking in background + ctx := context.Background() + go manager.StartHealthChecking(ctx) + + return &SMDPHandler{ + manager: manager, + } +} + +// AddCarrier adds a new carrier to the SM-DP+ manager +func (h *SMDPHandler) AddCarrier(c *gin.Context) { + var carrierConfig smdp.CarrierConfig + if err := c.ShouldBindJSON(&carrierConfig); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + carrier := carrierConfig.ToCarrier() + if err := h.manager.AddCarrier(carrier); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + + c.JSON(http.StatusCreated, gin.H{ + "message": "Carrier added successfully", + "carrier_id": carrier.ID, + }) +} + +// RemoveCarrier removes a carrier from the SM-DP+ manager +func (h *SMDPHandler) RemoveCarrier(c *gin.Context) { + carrierID := c.Param("id") + if carrierID == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "Carrier ID is required"}) + return + } + + if err := h.manager.RemoveCarrier(carrierID); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + + c.JSON(http.StatusOK, gin.H{"message": "Carrier removed successfully"}) +} + +// DownloadProfile handles eSIM profile download with multi-carrier support +func (h *SMDPHandler) DownloadProfile(c *gin.Context) { + var req smdp.ProfileRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) + defer cancel() + + response, err := h.manager.DownloadProfile(ctx, &req) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{ + "error": err.Error(), + "response": response, + }) + return + } + + c.JSON(http.StatusOK, response) +} + +// GetCarrierStatus returns the status of all carriers +func (h *SMDPHandler) GetCarrierStatus(c *gin.Context) { + status := h.manager.GetCarrierStatus() + c.JSON(http.StatusOK, gin.H{ + "carriers": status, + "timestamp": time.Now(), + }) +} + +// GetProfileStatus gets profile status from the best available carrier +func (h *SMDPHandler) GetProfileStatus(c *gin.Context) { + var req struct { + EID string `json:"eid" binding:"required"` + ICCID string `json:"iccid" binding:"required"` + } + + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + // Create profile request for status check + profileReq := &smdp.ProfileRequest{ + EID: req.EID, + ICCID: req.ICCID, + } + + ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) + defer cancel() + + // Use the manager to select best carrier and get status + carrier, err := h.manager.SelectCarrier(ctx, profileReq) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + + c.JSON(http.StatusOK, gin.H{ + "carrier_id": carrier.ID, + "carrier_name": carrier.Name, + "health_status": carrier.HealthStatus, + "message": "Use specific carrier endpoint for detailed status", + }) +} diff --git a/apps/carrier-connector/internal/handler/types.go b/apps/carrier-connector/internal/handlers/types.go similarity index 100% rename from apps/carrier-connector/internal/handler/types.go rename to apps/carrier-connector/internal/handlers/types.go diff --git a/apps/carrier-connector/internal/integration/smdp_integration.go b/apps/carrier-connector/internal/integration/smdp_integration.go new file mode 100644 index 0000000..634760b --- /dev/null +++ b/apps/carrier-connector/internal/integration/smdp_integration.go @@ -0,0 +1,198 @@ +package integration + +import ( + "fmt" + + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/handlers" + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository" + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/service" + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/smdp" + "github.com/gin-gonic/gin" + "github.com/sirupsen/logrus" +) + +// SMDPIntegration integrates the multi-carrier SM-DP+ system with the existing carrier connector +type SMDPIntegration struct { + service *service.SMDPService + handler *handlers.SMDPHandler + repository *repository.PostgresProfileStore + logger *logrus.Logger +} + +// NewSMDPIntegration creates a new SM-DP+ integration +func NewSMDPIntegration(repo *repository.PostgresProfileStore) *SMDPIntegration { + logger := logrus.New() + logger.SetLevel(logrus.InfoLevel) + + svc := service.NewSMDPService(repo) + hnd := handlers.NewSMDPHandler(repo) + + integration := &SMDPIntegration{ + service: svc, + handler: hnd, + repository: repo, + logger: logger, + } + + // Initialize default carriers + if err := integration.InitializeSystem(); err != nil { + logger.WithError(err).Error("Failed to initialize SM-DP+ integration") + } + + return integration +} + +// InitializeSystem sets up the SM-DP+ system with default carriers and configuration +func (i *SMDPIntegration) InitializeSystem() error { + i.logger.Info("Initializing SM-DP+ Multi-Carrier Integration") + + // Initialize default carriers + if err := i.service.InitializeDefaultCarriers(); err != nil { + return fmt.Errorf("failed to initialize default carriers: %w", err) + } + + // Log system status + carriers := i.service.GetCarrierHealth() + i.logger.WithField("carrier_count", len(carriers)).Info("SM-DP+ system initialized") + + for id, carrier := range carriers { + i.logger.WithFields(logrus.Fields{ + "carrier_id": id, + "carrier_name": carrier.Name, + "country": carrier.CountryCode, + "health_status": carrier.HealthStatus, + "is_active": carrier.IsActive, + }).Info("Carrier loaded") + } + + return nil +} + +// RegisterRoutes registers SM-DP+ routes with the Gin router +func (i *SMDPIntegration) RegisterRoutes(router *gin.RouterGroup) { + smdp := router.Group("/smdp") + { + // Profile management + smdp.POST("/download", i.handler.DownloadProfile) + smdp.POST("/status", i.handler.GetProfileStatus) + + // Carrier management + smdp.POST("/carriers", i.handler.AddCarrier) + smdp.DELETE("/carriers/:id", i.handler.RemoveCarrier) + smdp.GET("/carriers/status", i.handler.GetCarrierStatus) + + // Health and metrics + smdp.GET("/health", i.healthHandler) + smdp.GET("/metrics", i.metricsHandler) + } + + i.logger.Info("SM-DP+ routes registered") +} + +// healthHandler returns system health status +func (i *SMDPIntegration) healthHandler(c *gin.Context) { + carriers := i.service.GetCarrierHealth() + + healthyCount := 0 + totalCount := len(carriers) + + for _, carrier := range carriers { + if carrier.HealthStatus == smdp.CarrierStatusHealthy { + healthyCount++ + } + } + + status := "healthy" + if healthyCount == 0 { + status = "unhealthy" + } else if healthyCount < totalCount { + status = "degraded" + } + + c.JSON(200, gin.H{ + "status": status, + "healthy_carriers": healthyCount, + "total_carriers": totalCount, + "carriers": carriers, + }) +} + +// metricsHandler returns system metrics +func (i *SMDPIntegration) metricsHandler(c *gin.Context) { + metrics := i.service.GetCarrierMetrics() + carriers := i.service.GetCarrierHealth() + + response := gin.H{ + "carrier_metrics": metrics, + "carrier_details": carriers, + "timestamp": gin.H{}, + } + + // Calculate total metrics + var totalRequests, totalSuccess, totalFailed uint64 + for _, metric := range metrics { + totalRequests += metric.TotalRequests + totalSuccess += metric.SuccessfulRequests + totalFailed += metric.FailedRequests + } + + response["total_metrics"] = gin.H{ + "total_requests": totalRequests, + "successful_requests": totalSuccess, + "failed_requests": totalFailed, + "success_rate": float64(totalSuccess) / float64(totalRequests), + } + + c.JSON(200, response) +} + +// GetService returns the SM-DP+ service for direct access +func (i *SMDPIntegration) GetService() *service.SMDPService { + return i.service +} + +// GetHandler returns the SM-DP+ handler for direct access +func (i *SMDPIntegration) GetHandler() *handlers.SMDPHandler { + return i.handler +} + +// ExampleUsage demonstrates how to use the SM-DP+ integration +func ExampleUsage() { + // This would typically be in your main.go or router setup + + /* + // Initialize repository + repo, err := repository.NewPostgresProfileStore("postgres://user:pass@localhost/db") + if err != nil { + log.Fatal(err) + } + defer repo.Close() + + // Create SM-DP+ integration + smdpIntegration := integration.NewSMDPIntegration(repo) + + // Register routes + router := gin.Default() + api := router.Group("/api/v1") + smdpIntegration.RegisterRoutes(api) + + // Get service for direct use + service := smdpIntegration.GetService() + + // Example: Download profile + ctx := context.Background() + req := &smdp.ProfileRequest{ + EID: "eid-example", + ICCID: "iccid-example", + ProfileType: "operational", + } + + response, err := service.DownloadProfile(ctx, req) + if err != nil { + log.Printf("Profile download failed: %v", err) + return + } + + log.Printf("Profile downloaded successfully from carrier %s", response.CarrierID) + */ +} diff --git a/apps/carrier-connector/internal/service/smdp_service.go b/apps/carrier-connector/internal/service/smdp_service.go new file mode 100644 index 0000000..9da1ec9 --- /dev/null +++ b/apps/carrier-connector/internal/service/smdp_service.go @@ -0,0 +1,157 @@ +package service + +import ( + "context" + "fmt" + + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository" + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/smdp" + "github.com/sirupsen/logrus" +) + +// SMDPService provides high-level SM-DP+ operations with multi-carrier support +type SMDPService struct { + manager *smdp.SMDPManager + repository *repository.PostgresProfileStore + logger *logrus.Logger +} + +// NewSMDPService creates a new SM-DP+ service +func NewSMDPService(repo *repository.PostgresProfileStore) *SMDPService { + config := smdp.DefaultManagerConfig() + manager := smdp.NewSMDPManager(repo, config) + + logger := logrus.New() + logger.SetLevel(logrus.InfoLevel) + + // Start health checking + ctx := context.Background() + go manager.StartHealthChecking(ctx) + + return &SMDPService{ + manager: manager, + repository: repo, + logger: logger, + } +} + +// InitializeDefaultCarriers adds common carriers to the manager +func (s *SMDPService) InitializeDefaultCarriers() error { + defaultCarriers := []*smdp.CarrierConfig{ + { + ID: "carrier-att-us", + Name: "AT&T US", + CountryCode: "US", + MCC: "310", + MNC: "410", + ES2BaseURL: "https://smdp.att.com", + ES2APIKey: "", + ES2InsecureSkip: false, + Priority: 100, + IsActive: true, + MaxConcurrentReqs: 100, + SupportedProfileTypes: []string{"operational", "test"}, + SupportedMCCs: []string{"310", "311", "312", "313", "314", "315", "316"}, + Features: []string{"bulk_download", "remote_provisioning"}, + }, + { + ID: "carrier-verizon-us", + Name: "Verizon US", + CountryCode: "US", + MCC: "311", + MNC: "480", + ES2BaseURL: "https://smdp.verizon.com", + ES2APIKey: "", + ES2InsecureSkip: false, + Priority: 90, + IsActive: true, + MaxConcurrentReqs: 80, + SupportedProfileTypes: []string{"operational", "test"}, + SupportedMCCs: []string{"311", "312"}, + Features: []string{"bulk_download"}, + }, + { + ID: "carrier-tmobile-de", + Name: "T-Mobile DE", + CountryCode: "DE", + MCC: "262", + MNC: "01", + ES2BaseURL: "https://smdp.t-mobile.de", + ES2APIKey: "", + ES2InsecureSkip: false, + Priority: 85, + IsActive: true, + MaxConcurrentReqs: 60, + SupportedProfileTypes: []string{"operational"}, + SupportedMCCs: []string{"262"}, + Features: []string{"bulk_download", "remote_provisioning"}, + }, + } + + for _, carrierConfig := range defaultCarriers { + carrier := carrierConfig.ToCarrier() + if err := s.manager.AddCarrier(carrier); err != nil { + s.logger.WithError(err).WithField("carrier_id", carrier.ID). + Warn("Failed to add default carrier") + continue + } + s.logger.WithField("carrier_id", carrier.ID).Info("Added default carrier") + } + + return nil +} + +// DownloadProfile handles profile download with intelligent carrier selection +func (s *SMDPService) DownloadProfile(ctx context.Context, req *smdp.ProfileRequest) (*smdp.ProfileResponse, error) { + s.logger.WithFields(logrus.Fields{ + "eid": req.EID, + "iccid": req.ICCID, + "preferred": req.PreferredCarrier, + }).Info("Processing profile download through SM-DP+ service") + + return s.manager.DownloadProfile(ctx, req) +} + +// GetOptimalCarrier selects the best carrier for a given request +func (s *SMDPService) GetOptimalCarrier(ctx context.Context, req *smdp.ProfileRequest) (*smdp.Carrier, error) { + return s.manager.SelectCarrier(ctx, req) +} + +// GetCarrierHealth returns health status of all carriers +func (s *SMDPService) GetCarrierHealth() map[string]*smdp.Carrier { + return s.manager.GetCarrierStatus() +} + +// AddCarrier dynamically adds a new carrier +func (s *SMDPService) AddCarrier(carrierConfig *smdp.CarrierConfig) error { + carrier := carrierConfig.ToCarrier() + + if err := s.manager.AddCarrier(carrier); err != nil { + return fmt.Errorf("failed to add carrier %s: %w", carrier.ID, err) + } + + s.logger.WithField("carrier_id", carrier.ID).Info("Successfully added new carrier") + return nil +} + +// RemoveCarrier removes a carrier from the system +func (s *SMDPService) RemoveCarrier(carrierID string) error { + if err := s.manager.RemoveCarrier(carrierID); err != nil { + return fmt.Errorf("failed to remove carrier %s: %w", carrierID, err) + } + + s.logger.WithField("carrier_id", carrierID).Info("Successfully removed carrier") + return nil +} + +// GetCarrierMetrics returns performance metrics for all carriers +func (s *SMDPService) GetCarrierMetrics() map[string]*smdp.CarrierMetrics { + carriers := s.manager.GetCarrierStatus() + metrics := make(map[string]*smdp.CarrierMetrics) + + for id, carrier := range carriers { + metrics[id] = carrier.Metrics + } + + return metrics +} diff --git a/apps/carrier-connector/internal/smdp/config.go b/apps/carrier-connector/internal/smdp/config.go new file mode 100644 index 0000000..a95d4ee --- /dev/null +++ b/apps/carrier-connector/internal/smdp/config.go @@ -0,0 +1,77 @@ +package smdp + +import ( + "time" + + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/config" +) + +// DefaultManagerConfig returns default configuration for SM-DP+ Manager +func DefaultManagerConfig() *ManagerConfig { + return &ManagerConfig{ + HealthCheckInterval: 30 * time.Second, + MaxRetries: 3, + RetryDelay: 2 * time.Second, + CircuitBreakerThreshold: 5, + CircuitBreakerTimeout: 60 * time.Second, + EnableLoadBalancing: true, + EnableFailover: true, + DefaultTimeout: 30 * time.Second, + } +} + +// CarrierConfig represents carrier configuration from database or config file +type CarrierConfig struct { + ID string `json:"id" yaml:"id"` + Name string `json:"name" yaml:"name"` + CountryCode string `json:"country_code" yaml:"country_code"` + MCC string `json:"mcc" yaml:"mcc"` + MNC string `json:"mnc" yaml:"mnc"` + ES2BaseURL string `json:"es2_base_url" yaml:"es2_base_url"` + ES2APIKey string `json:"es2_api_key" yaml:"es2_api_key"` + ES2InsecureSkip bool `json:"es2_insecure_skip" yaml:"es2_insecure_skip"` + Priority int `json:"priority" yaml:"priority"` + IsActive bool `json:"is_active" yaml:"is_active"` + MaxConcurrentReqs int `json:"max_concurrent_requests" yaml:"max_concurrent_requests"` + SupportedProfileTypes []string `json:"supported_profile_types" yaml:"supported_profile_types"` + SupportedMCCs []string `json:"supported_mccs" yaml:"supported_mccs"` + SupportedMNCs []string `json:"supported_mncs" yaml:"supported_mncs"` + Features []string `json:"features" yaml:"features"` +} + +// ToCarrier converts CarrierConfig to Carrier struct +func (c *CarrierConfig) ToCarrier() *Carrier { + carrier := &Carrier{ + ID: c.ID, + Name: c.Name, + CountryCode: c.CountryCode, + MCC: c.MCC, + MNC: c.MNC, + Priority: c.Priority, + IsActive: c.IsActive, + HealthStatus: CarrierStatusUnknown, + LastHealthCheck: time.Now(), + Metrics: &CarrierMetrics{ + TotalRequests: 0, + SuccessfulRequests: 0, + FailedRequests: 0, + AverageResponseTime: 0, + RequestRate: 0, + }, + Capabilities: &CarrierCapabilities{ + SupportedProfileTypes: c.SupportedProfileTypes, + MaxConcurrentRequests: c.MaxConcurrentReqs, + SupportedMCCs: c.SupportedMCCs, + SupportedMNCs: c.SupportedMNCs, + Features: c.Features, + }, + ES2Config: &config.ES2Config{ + BaseURL: c.ES2BaseURL, + APIKey: c.ES2APIKey, + InsecureSkipVerify: c.ES2InsecureSkip, + FunctionalityRequesterID: "telecom-platform", + }, + } + + return carrier +} diff --git a/apps/carrier-connector/internal/smdp/health_checker.go b/apps/carrier-connector/internal/smdp/health_checker.go new file mode 100644 index 0000000..4eda1b6 --- /dev/null +++ b/apps/carrier-connector/internal/smdp/health_checker.go @@ -0,0 +1,116 @@ +package smdp + +import ( + "context" + "sync" + "time" + + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2" + "github.com/sirupsen/logrus" +) + +// HealthChecker performs periodic health checks on carriers +type HealthChecker struct { + interval time.Duration + logger *logrus.Logger + healthUpdateFn func(carrierID string, status CarrierHealthStatus) +} + +// NewHealthChecker creates a new health checker +func NewHealthChecker(interval time.Duration) *HealthChecker { + return &HealthChecker{ + interval: interval, + logger: logrus.New(), + } +} + +// Start begins the health checking process +func (h *HealthChecker) Start(ctx context.Context, carriers map[string]*Carrier, clients map[string]*es2.ES2Client, updateFn func(string, CarrierHealthStatus)) { + h.healthUpdateFn = updateFn + + ticker := time.NewTicker(h.interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + h.logger.Info("Health checker stopped") + return + case <-ticker.C: + h.checkAllCarriers(carriers, clients) + } + } +} + +// checkAllCarriers performs health checks on all carriers +func (h *HealthChecker) checkAllCarriers(carriers map[string]*Carrier, clients map[string]*es2.ES2Client) { + var wg sync.WaitGroup + + for carrierID, carrier := range carriers { + if !carrier.IsActive { + continue + } + + wg.Add(1) + go func(id string, c *Carrier) { + defer wg.Done() + h.checkCarrier(id, c, clients[id]) + }(carrierID, carrier) + } + + wg.Wait() +} + +// checkCarrier performs a health check on a single carrier +func (h *HealthChecker) checkCarrier(carrierID string, carrier *Carrier, client *es2.ES2Client) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + startTime := time.Now() + + // Perform a simple health check by attempting to get profile status + // This is a lightweight operation that tests connectivity + req := &es2.GetProfileStatusRequest{ + EID: "health-check-eid", + ICCID: "health-check-iccid", + } + + _, err := client.GetProfileStatus(ctx, req) + responseTime := time.Since(startTime) + + status := h.evaluateHealth(err, responseTime, carrier.Metrics) + h.healthUpdateFn(carrierID, status) + + h.logger.WithFields(logrus.Fields{ + "carrier_id": carrierID, + "status": status, + "response_time": responseTime, + "error": err, + }).Debug("Health check completed") +} + +// evaluateHealth evaluates the health status based on error and response time +func (h *HealthChecker) evaluateHealth(err error, responseTime time.Duration, metrics *CarrierMetrics) CarrierHealthStatus { + if err != nil { + // Check if this is the first failure or consecutive failures + if metrics.FailedRequests > 5 { + return CarrierStatusUnhealthy + } + return CarrierStatusDegraded + } + + // Check response time + if responseTime > 5*time.Second { + return CarrierStatusDegraded + } + + // Check error rate + if metrics.TotalRequests > 0 { + errorRate := float64(metrics.FailedRequests) / float64(metrics.TotalRequests) + if errorRate > 0.1 { // More than 10% error rate + return CarrierStatusDegraded + } + } + + return CarrierStatusHealthy +} diff --git a/apps/carrier-connector/internal/smdp/load_balancer.go b/apps/carrier-connector/internal/smdp/load_balancer.go new file mode 100644 index 0000000..6af6b6a --- /dev/null +++ b/apps/carrier-connector/internal/smdp/load_balancer.go @@ -0,0 +1,133 @@ +package smdp + +import ( + "fmt" + "math/rand" + "time" +) + +// LoadBalancer implements different load balancing strategies for carrier selection +type LoadBalancer struct { + strategy LoadBalancingStrategy + rand *rand.Rand +} + +// LoadBalancingStrategy defines the load balancing algorithm +type LoadBalancingStrategy int + +const ( + StrategyRoundRobin LoadBalancingStrategy = iota + StrategyWeightedRoundRobin + StrategyLeastConnections + StrategyRandom + StrategyPriority +) + +// NewLoadBalancer creates a new load balancer with default strategy +func NewLoadBalancer() *LoadBalancer { + return &LoadBalancer{ + strategy: StrategyWeightedRoundRobin, + rand: rand.New(rand.NewSource(time.Now().UnixNano())), + } +} + +// SelectCarrier selects the best carrier based on the configured strategy +func (lb *LoadBalancer) SelectCarrier(carriers []*Carrier, req *ProfileRequest) (*Carrier, error) { + if len(carriers) == 0 { + return nil, fmt.Errorf("no carriers available") + } + + switch lb.strategy { + case StrategyRoundRobin: + return lb.roundRobinSelect(carriers), nil + case StrategyWeightedRoundRobin: + return lb.weightedRoundRobinSelect(carriers), nil + case StrategyLeastConnections: + return lb.leastConnectionsSelect(carriers), nil + case StrategyRandom: + return lb.randomSelect(carriers), nil + case StrategyPriority: + return lb.prioritySelect(carriers), nil + default: + return lb.weightedRoundRobinSelect(carriers), nil + } +} + +// roundRobinSelect implements round-robin carrier selection +func (lb *LoadBalancer) roundRobinSelect(carriers []*Carrier) *Carrier { + // Simple round-robin based on request count + // In a real implementation, you'd maintain state across requests + return carriers[0] // Simplified for demo +} + +// weightedRoundRobinSelect selects carrier based on priority and health +func (lb *LoadBalancer) weightedRoundRobinSelect(carriers []*Carrier) *Carrier { + // Calculate weights based on priority and success rate + var totalWeight int + weights := make([]int, len(carriers)) + + for i, carrier := range carriers { + weight := carrier.Priority + + // Adjust weight based on success rate + if carrier.Metrics.TotalRequests > 0 { + successRate := float64(carrier.Metrics.SuccessfulRequests) / float64(carrier.Metrics.TotalRequests) + weight = int(float64(weight) * successRate) + } + + weights[i] = weight + totalWeight += weight + } + + if totalWeight == 0 { + return carriers[0] + } + + // Select carrier based on weighted random + random := lb.rand.Intn(totalWeight) + currentWeight := 0 + + for i, weight := range weights { + currentWeight += weight + if random < currentWeight { + return carriers[i] + } + } + + return carriers[0] +} + +// leastConnectionsSelect selects carrier with least current load +func (lb *LoadBalancer) leastConnectionsSelect(carriers []*Carrier) *Carrier { + bestCarrier := carriers[0] + minLoad := bestCarrier.Metrics.RequestRate + + for _, carrier := range carriers[1:] { + if carrier.Metrics.RequestRate < minLoad { + bestCarrier = carrier + minLoad = carrier.Metrics.RequestRate + } + } + + return bestCarrier +} + +// randomSelect selects a random carrier +func (lb *LoadBalancer) randomSelect(carriers []*Carrier) *Carrier { + return carriers[lb.rand.Intn(len(carriers))] +} + +// prioritySelect selects carrier with highest priority +func (lb *LoadBalancer) prioritySelect(carriers []*Carrier) *Carrier { + bestCarrier := carriers[0] + highestPriority := bestCarrier.Priority + + for _, carrier := range carriers[1:] { + if carrier.Priority > highestPriority { + bestCarrier = carrier + highestPriority = carrier.Priority + } + } + + return bestCarrier +} diff --git a/apps/carrier-connector/internal/smdp/manager.go b/apps/carrier-connector/internal/smdp/manager.go new file mode 100644 index 0000000..3b97b0e --- /dev/null +++ b/apps/carrier-connector/internal/smdp/manager.go @@ -0,0 +1,121 @@ +package smdp + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2" + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository" + "github.com/sirupsen/logrus" +) + +// SMDPManager manages multiple SM-DP+ carriers +type SMDPManager struct { + carriers map[string]*Carrier + carriersMutex sync.RWMutex + es2Clients map[string]*es2.ES2Client + clientsMutex sync.RWMutex + repository *repository.PostgresProfileStore + healthChecker *HealthChecker + loadBalancer *LoadBalancer + config *ManagerConfig + logger *logrus.Logger +} + +// NewSMDPManager creates a new SM-DP+ Manager +func NewSMDPManager(repo *repository.PostgresProfileStore, config *ManagerConfig) *SMDPManager { + logger := logrus.New() + logger.SetLevel(logrus.InfoLevel) + + return &SMDPManager{ + carriers: make(map[string]*Carrier), + es2Clients: make(map[string]*es2.ES2Client), + repository: repo, + config: config, + logger: logger, + healthChecker: NewHealthChecker(config.HealthCheckInterval), + loadBalancer: NewLoadBalancer(), + } +} + +// AddCarrier adds a new carrier to the manager +func (m *SMDPManager) AddCarrier(carrier *Carrier) error { + m.carriersMutex.Lock() + defer m.carriersMutex.Unlock() + + if err := m.validateCarrier(carrier); err != nil { + return fmt.Errorf("invalid carrier configuration: %w", err) + } + + client := es2.NewES2Client(carrier.ES2Config) + m.carriers[carrier.ID] = carrier + m.es2Clients[carrier.ID] = client + + m.logger.WithFields(logrus.Fields{ + "carrier_id": carrier.ID, + "carrier_name": carrier.Name, + }).Info("Added carrier to SM-DP+ Manager") + + return nil +} + +// RemoveCarrier removes a carrier from the manager +func (m *SMDPManager) RemoveCarrier(carrierID string) error { + m.carriersMutex.Lock() + defer m.carriersMutex.Unlock() + + m.clientsMutex.Lock() + defer m.clientsMutex.Unlock() + + if _, exists := m.carriers[carrierID]; !exists { + return fmt.Errorf("carrier %s not found", carrierID) + } + + delete(m.carriers, carrierID) + delete(m.es2Clients, carrierID) + + m.logger.WithField("carrier_id", carrierID).Info("Removed carrier from SM-DP+ Manager") + return nil +} + +// GetCarrierStatus returns the current status of all carriers +func (m *SMDPManager) GetCarrierStatus() map[string]*Carrier { + m.carriersMutex.RLock() + defer m.carriersMutex.RUnlock() + + status := make(map[string]*Carrier) + for id, carrier := range m.carriers { + carrierCopy := *carrier + status[id] = &carrierCopy + } + return status +} + +// StartHealthChecking starts the background health checking process +func (m *SMDPManager) StartHealthChecking(ctx context.Context) { + m.healthChecker.Start(ctx, m.carriers, m.es2Clients, m.updateCarrierHealth) +} + +// validateCarrier validates carrier configuration +func (m *SMDPManager) validateCarrier(carrier *Carrier) error { + if carrier.ID == "" || carrier.Name == "" || carrier.MCC == "" || carrier.MNC == "" { + return fmt.Errorf("carrier ID, name, MCC, and MNC are required") + } + if carrier.ES2Config == nil || carrier.ES2Config.BaseURL == "" { + return fmt.Errorf("ES2+ configuration and base URL are required") + } + return nil +} + +// updateCarrierHealth updates the health status of a carrier +func (m *SMDPManager) updateCarrierHealth(carrierID string, status CarrierHealthStatus) { + m.carriersMutex.Lock() + defer m.carriersMutex.Unlock() + + if carrier, exists := m.carriers[carrierID]; exists { + carrier.HealthStatus = status + carrier.LastHealthCheck = time.Now() + } +} diff --git a/apps/carrier-connector/internal/smdp/operations.go b/apps/carrier-connector/internal/smdp/operations.go new file mode 100644 index 0000000..e7175ad --- /dev/null +++ b/apps/carrier-connector/internal/smdp/operations.go @@ -0,0 +1,200 @@ +package smdp + +import ( + "context" + "fmt" + "time" + + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2" + "github.com/sirupsen/logrus" +) + +func (m *SMDPManager) DownloadProfile(ctx context.Context, req *ProfileRequest) (*ProfileResponse, error) { + startTime := time.Now() + + m.logger.WithFields(logrus.Fields{ + "eid": req.EID, + "preferred": req.PreferredCarrier, + }).Info("Processing profile download request") + + selectedCarrier, err := m.SelectCarrier(ctx, req) + if err != nil { + return &ProfileResponse{ + Success: false, + StatusMessage: fmt.Sprintf("Carrier selection failed: %v", err), + ResponseTime: time.Since(startTime), + }, err + } + + response, err := m.downloadWithRetry(ctx, req, selectedCarrier, 0) + if err != nil && m.config.EnableFailover { + m.logger.WithError(err).Warn("Primary carrier failed, attempting failover") + return m.handleFailover(ctx, req, selectedCarrier.ID, startTime) + } + + response.ResponseTime = time.Since(startTime) + return response, err +} + +func (m *SMDPManager) SelectCarrier(ctx context.Context, req *ProfileRequest) (*Carrier, error) { + m.carriersMutex.RLock() + defer m.carriersMutex.RUnlock() + + var activeCarriers []*Carrier + for _, carrier := range m.carriers { + if carrier.IsActive && carrier.HealthStatus == CarrierStatusHealthy { + activeCarriers = append(activeCarriers, carrier) + } + } + + if len(activeCarriers) == 0 { + return nil, fmt.Errorf("no healthy carriers available") + } + + if req.PreferredCarrier != "" { + if carrier, exists := m.carriers[req.PreferredCarrier]; exists && + carrier.IsActive && carrier.HealthStatus == CarrierStatusHealthy { + return carrier, nil + } + } + + if m.config.EnableLoadBalancing { + return m.loadBalancer.SelectCarrier(activeCarriers, req) + } + + return m.getHighestPriorityCarrier(activeCarriers), nil +} + +func (m *SMDPManager) downloadWithRetry(ctx context.Context, req *ProfileRequest, carrier *Carrier, attempt int) (*ProfileResponse, error) { + m.clientsMutex.RLock() + client, exists := m.es2Clients[carrier.ID] + m.clientsMutex.RUnlock() + + if !exists { + return nil, fmt.Errorf("ES2+ client not found for carrier %s", carrier.ID) + } + + downloadReq := &es2.DownloadProfileRequest{ + EID: req.EID, + ICCID: req.ICCID, + ProfileType: req.ProfileType, + ConfirmationCode: req.ConfirmationCode, + } + + startTime := time.Now() + resp, err := client.DownloadProfile(ctx, downloadReq) + responseTime := time.Since(startTime) + + m.updateCarrierMetrics(carrier.ID, err == nil, responseTime, err) + + if err != nil { + m.logger.WithFields(logrus.Fields{ + "carrier_id": carrier.ID, + "attempt": attempt + 1, + "error": err, + }).Error("Profile download failed") + + if attempt < m.config.MaxRetries { + time.Sleep(m.config.RetryDelay) + return m.downloadWithRetry(ctx, req, carrier, attempt+1) + } + + return &ProfileResponse{ + Success: false, + CarrierID: carrier.ID, + StatusMessage: fmt.Sprintf("Download failed after %d attempts: %v", attempt+1, err), + ResponseTime: responseTime, + }, err + } + + m.logger.WithFields(logrus.Fields{ + "carrier_id": carrier.ID, + "execution_status": resp.ExecutionStatus, + "response_time": responseTime, + }).Info("Profile download successful") + + return &ProfileResponse{ + Success: true, + CarrierID: carrier.ID, + ExecutionStatus: resp.ExecutionStatus, + StatusMessage: resp.StatusMessage, + ResponseTime: responseTime, + }, nil +} + +// handleFailover attempts to download profile using alternative carriers +func (m *SMDPManager) handleFailover(ctx context.Context, req *ProfileRequest, failedCarrierID string, startTime time.Time) (*ProfileResponse, error) { + m.carriersMutex.RLock() + defer m.carriersMutex.RUnlock() + + var retriedOn []string + + for carrierID, carrier := range m.carriers { + if carrierID == failedCarrierID || !carrier.IsActive || carrier.HealthStatus != CarrierStatusHealthy { + continue + } + + m.logger.WithField("carrier_id", carrierID).Info("Attempting failover to carrier") + + response, err := m.downloadWithRetry(ctx, req, carrier, 0) + if err == nil && response.Success { + response.RetriedOn = retriedOn + response.ResponseTime = time.Since(startTime) + return response, nil + } + + retriedOn = append(retriedOn, carrierID) + } + + return &ProfileResponse{ + Success: false, + StatusMessage: "All carriers failed during failover", + ResponseTime: time.Since(startTime), + RetriedOn: retriedOn, + }, fmt.Errorf("failover exhausted all carriers") +} + +// updateCarrierMetrics updates the metrics for a carrier +func (m *SMDPManager) updateCarrierMetrics(carrierID string, success bool, responseTime time.Duration, err error) { + m.carriersMutex.Lock() + defer m.carriersMutex.Unlock() + + carrier, exists := m.carriers[carrierID] + if !exists { + return + } + + carrier.Metrics.TotalRequests++ + if success { + carrier.Metrics.SuccessfulRequests++ + } else { + carrier.Metrics.FailedRequests++ + carrier.Metrics.LastError = err.Error() + carrier.Metrics.LastErrorTime = time.Now() + } + + if carrier.Metrics.TotalRequests == 1 { + carrier.Metrics.AverageResponseTime = responseTime + } else { + carrier.Metrics.AverageResponseTime = time.Duration( + (int64(carrier.Metrics.AverageResponseTime)*int64(carrier.Metrics.TotalRequests-1) + int64(responseTime)) / int64(carrier.Metrics.TotalRequests), + ) + } + + carrier.Metrics.RequestRate = float64(carrier.Metrics.TotalRequests) / time.Since(time.Now().Add(-time.Minute)).Seconds() +} + +func (m *SMDPManager) getHighestPriorityCarrier(carriers []*Carrier) *Carrier { + if len(carriers) == 0 { + return nil + } + + highestPriority := carriers[0] + for _, carrier := range carriers { + if carrier.Priority > highestPriority.Priority { + highestPriority = carrier + } + } + + return highestPriority +} diff --git a/apps/carrier-connector/internal/smdp/types.go b/apps/carrier-connector/internal/smdp/types.go new file mode 100644 index 0000000..0a1835e --- /dev/null +++ b/apps/carrier-connector/internal/smdp/types.go @@ -0,0 +1,87 @@ +package smdp + +import ( + "time" + + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/config" +) + +// Carrier represents a mobile network operator with SM-DP+ capabilities +type Carrier struct { + ID string `json:"id"` + Name string `json:"name"` + CountryCode string `json:"country_code"` + MCC string `json:"mcc"` + MNC string `json:"mnc"` + ES2Config *config.ES2Config `json:"es2_config"` + Priority int `json:"priority"` // Higher number = higher priority + IsActive bool `json:"is_active"` + HealthStatus CarrierHealthStatus `json:"health_status"` + LastHealthCheck time.Time `json:"last_health_check"` + Metrics *CarrierMetrics `json:"metrics"` + Capabilities *CarrierCapabilities `json:"capabilities"` +} + +// CarrierHealthStatus represents the health status of a carrier +type CarrierHealthStatus string + +const ( + CarrierStatusHealthy CarrierHealthStatus = "healthy" + CarrierStatusDegraded CarrierHealthStatus = "degraded" + CarrierStatusUnhealthy CarrierHealthStatus = "unhealthy" + CarrierStatusUnknown CarrierHealthStatus = "unknown" +) + +// CarrierCapabilities represents the capabilities of a carrier +type CarrierCapabilities struct { + SupportedProfileTypes []string `json:"supported_profile_types"` + MaxConcurrentRequests int `json:"max_concurrent_requests"` + SupportedMCCs []string `json:"supported_mccs"` + SupportedMNCs []string `json:"supported_mncs"` + Features []string `json:"features"` // e.g., ["bulk_download", "remote_provisioning"] +} + +// CarrierMetrics tracks performance metrics for a carrier +type CarrierMetrics struct { + TotalRequests uint64 `json:"total_requests"` + SuccessfulRequests uint64 `json:"successful_requests"` + FailedRequests uint64 `json:"failed_requests"` + AverageResponseTime time.Duration `json:"average_response_time"` + LastError string `json:"last_error"` + LastErrorTime time.Time `json:"last_error_time"` + RequestRate float64 `json:"request_rate"` // requests per second +} + +// ProfileRequest represents an eSIM profile download request +type ProfileRequest struct { + EID string `json:"eid"` + ICCID string `json:"iccid"` + ProfileType string `json:"profile_type"` + ConfirmationCode string `json:"confirmation_code,omitempty"` + PreferredCarrier string `json:"preferred_carrier,omitempty"` // Optional preferred carrier + IMSI string `json:"imsi,omitempty"` + TenantID string `json:"tenant_id,omitempty"` +} + +// ProfileResponse represents the response from a profile operation +type ProfileResponse struct { + Success bool `json:"success"` + CarrierID string `json:"carrier_id"` + ExecutionStatus string `json:"execution_status"` + StatusMessage string `json:"status_message"` + ProfileState string `json:"profile_state,omitempty"` + ResponseTime time.Duration `json:"response_time"` + RetriedOn []string `json:"retried_on,omitempty"` // List of carrier IDs tried before success +} + +// ManagerConfig configures the SM-DP+ Manager +type ManagerConfig struct { + HealthCheckInterval time.Duration `json:"health_check_interval"` + MaxRetries int `json:"max_retries"` + RetryDelay time.Duration `json:"retry_delay"` + CircuitBreakerThreshold int `json:"circuit_breaker_threshold"` + CircuitBreakerTimeout time.Duration `json:"circuit_breaker_timeout"` + EnableLoadBalancing bool `json:"enable_load_balancing"` + EnableFailover bool `json:"enable_failover"` + DefaultTimeout time.Duration `json:"default_timeout"` +} diff --git a/apps/carrier-connector/main.go b/apps/carrier-connector/main.go index 9671ee2..8d0c5f5 100644 --- a/apps/carrier-connector/main.go +++ b/apps/carrier-connector/main.go @@ -9,7 +9,7 @@ import ( "github.com/joho/godotenv" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/config" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2" - handler "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/handler" + handler "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/handlers" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/mq" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/webhook" diff --git a/apps/carrier-connector/routes.go b/apps/carrier-connector/routes.go index 5ecd492..aaaac80 100644 --- a/apps/carrier-connector/routes.go +++ b/apps/carrier-connector/routes.go @@ -8,7 +8,7 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2" - handler "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/handler" + handler "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/handlers" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/mq" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/webhook"