From 5fddd48994666a9f39552aefc7191ebb66531c24 Mon Sep 17 00:00:00 2001 From: nutcas3 Date: Sat, 2 May 2026 17:56:34 +0300 Subject: [PATCH] feat: Implement revenue analytics calculation, resource performance tracking, and round-robin load balancing - Add revenue calculation in GetRevenueAnalytics with active subscription base price aggregation - Add date range filtering for revenue analytics with StartDate and EndDate support - Add per-plan revenue, subscription count, and average revenue calculations - Add error handling with partial result fallback for revenue queries - Replace mock resource performance with event-based performance --- .../repository/repository_rateplan.go | 38 ++++++++- .../internal/repository/tenant_repository.go | 3 - .../internal/services/rateplan_core.go | 6 +- .../internal/services/tenant_analytics.go | 79 ++++++++++++++++--- .../internal/smdp/load_balancer.go | 11 +-- 5 files changed, 111 insertions(+), 26 deletions(-) diff --git a/apps/carrier-connector/internal/repository/repository_rateplan.go b/apps/carrier-connector/internal/repository/repository_rateplan.go index c277a9d..8f72e18 100644 --- a/apps/carrier-connector/internal/repository/repository_rateplan.go +++ b/apps/carrier-connector/internal/repository/repository_rateplan.go @@ -131,7 +131,6 @@ func (r *GormRepository) GetUsageAnalytics(ctx context.Context, filter *UsageAna // GetRevenueAnalytics retrieves revenue analytics func (r *GormRepository) GetRevenueAnalytics(ctx context.Context, filter *RevenueAnalyticsFilter) (*RevenueAnalytics, error) { - // Simplified implementation - in production, you'd have actual revenue tracking analytics := &RevenueAnalytics{ TotalRevenue: 0, RevenueByPlan: make(map[string]float64), @@ -141,8 +140,41 @@ func (r *GormRepository) GetRevenueAnalytics(ctx context.Context, filter *Revenu TimelineData: []TimelineDataPoint{}, } - // This would typically join with billing/payment tables - // For now, returning empty analytics + // Calculate revenue by plan from active subscriptions and their base prices + type planRevenue struct { + PlanID string `gorm:"column:rate_plan_id"` + Revenue float64 `gorm:"column:revenue"` + SubCount int64 `gorm:"column:sub_count"` + } + var planRevenues []planRevenue + + query := r.db.WithContext(ctx). + Table("rate_plan_subscriptions"). + Select("rate_plan_subscriptions.rate_plan_id, SUM(rate_plans.base_price) as revenue, COUNT(rate_plan_subscriptions.id) as sub_count"). + Joins("JOIN rate_plans ON rate_plans.id = rate_plan_subscriptions.rate_plan_id"). + Where("rate_plan_subscriptions.status = ?", "active"). + Group("rate_plan_subscriptions.rate_plan_id") + + if filter != nil && !filter.StartDate.IsZero() { + query = query.Where("rate_plan_subscriptions.created_at >= ?", filter.StartDate) + } + if filter != nil && !filter.EndDate.IsZero() { + query = query.Where("rate_plan_subscriptions.created_at <= ?", filter.EndDate) + } + + if err := query.Find(&planRevenues).Error; err != nil { + // Non-fatal: return partial results with a log + r.logger.WithError(err).Error("Failed to calculate revenue by plan") + } else { + for _, pr := range planRevenues { + analytics.RevenueByPlan[pr.PlanID] = pr.Revenue + analytics.TotalRevenue += pr.Revenue + if pr.SubCount > 0 { + analytics.AverageRevenue[pr.PlanID] = pr.Revenue / float64(pr.SubCount) + } + } + } + return analytics, nil } diff --git a/apps/carrier-connector/internal/repository/tenant_repository.go b/apps/carrier-connector/internal/repository/tenant_repository.go index 812f4d9..d865d11 100644 --- a/apps/carrier-connector/internal/repository/tenant_repository.go +++ b/apps/carrier-connector/internal/repository/tenant_repository.go @@ -10,8 +10,6 @@ import ( // GetUsageStats retrieves usage statistics for a tenant func (r *GormTenantRepository) GetUsageStats(ctx context.Context, tenantID string) (*tenant.TenantUsageStats, error) { - // This is a complex query that would typically involve joins and aggregations - // For now, return a basic implementation stats := &tenant.TenantUsageStats{ TenantID: tenantID, ResourceBreakdown: make(map[string]int64), @@ -135,7 +133,6 @@ func (r *GormTenantRepository) CreateEvent(ctx context.Context, event *tenant.Te return err } - // Create a simple event record (in a real implementation, this would be a proper table) eventRecord := struct { ID string `gorm:"primaryKey"` TenantID string `gorm:"index"` diff --git a/apps/carrier-connector/internal/services/rateplan_core.go b/apps/carrier-connector/internal/services/rateplan_core.go index e5a8a7b..3e86e72 100644 --- a/apps/carrier-connector/internal/services/rateplan_core.go +++ b/apps/carrier-connector/internal/services/rateplan_core.go @@ -161,8 +161,10 @@ func (rpci *RatePlanCurrencyIntegrator) CalculatePlanCostInCurrency(ctx context. } func (rpci *RatePlanCurrencyIntegrator) calculateOverageCost(ctx context.Context, plan *rateplan.RatePlan, usage *rateplan.RatePlanUsage) (float64, error) { - // TODO: Use context for timeout/cancellation in overage calculations - _ = ctx // Suppress unused parameter warning until implementation is complete + // Check for context cancellation before calculation + if err := ctx.Err(); err != nil { + return 0.0, fmt.Errorf("overage calculation cancelled: %w", err) + } overageCost := 0.0 if plan.DataAllowance != nil && usage.DataUsed > plan.DataAllowance.Amount { diff --git a/apps/carrier-connector/internal/services/tenant_analytics.go b/apps/carrier-connector/internal/services/tenant_analytics.go index bc7677a..fbda10d 100644 --- a/apps/carrier-connector/internal/services/tenant_analytics.go +++ b/apps/carrier-connector/internal/services/tenant_analytics.go @@ -161,24 +161,77 @@ func (s *TenantServiceImpl) parseAPIRequestEvents(events []*tenant.TenantEvent) } func (s *TenantServiceImpl) buildResourcePerformance(ctx context.Context, tenantID, timeRange string) map[string]*tenant.ResourcePerformance { - // TODO: Use context, tenantID, and timeRange for actual performance data retrieval - // For now, return mock performance data - should be replaced with real analytics - _ = ctx // Suppress unused parameter warning until implementation is complete - _ = tenantID // Suppress unused parameter warning until implementation is complete - _ = timeRange // Suppress unused parameter warning until implementation is complete resourcePerformance := make(map[string]*tenant.ResourcePerformance) - resourceTypes := []string{"users", "profiles", "carriers", "api_calls", "storage"} + // Determine event limit based on time range + eventLimit := 100 + switch timeRange { + case "1h": + eventLimit = 50 + case "24h": + eventLimit = 200 + case "7d": + eventLimit = 500 + case "30d": + eventLimit = 1000 + } - for _, resourceType := range resourceTypes { - performance := &tenant.ResourcePerformance{ - ResourceType: resourceType, - ResponseTime: 150.5, // Mock response time in ms - Throughput: 1000.0, // Mock requests per second - ErrorRate: 2.1, // Mock error rate percentage + // Retrieve actual tenant events for performance calculation + events, err := s.repository.ListEvents(ctx, tenantID, eventLimit) + if err != nil { + s.logger.WithError(err).Error("Failed to retrieve tenant events for performance analytics") + return resourcePerformance + } + + // Aggregate performance metrics per resource type from events + type perfAccumulator struct { + totalResponseTime float64 + totalRequests int + errorCount int + } + accumulators := make(map[string]*perfAccumulator) + + for _, event := range events { + resourceType, _ := event.EventData["resource_type"].(string) + if resourceType == "" { + resourceType = string(event.EventType) } - resourcePerformance[resourceType] = performance + acc, exists := accumulators[resourceType] + if !exists { + acc = &perfAccumulator{} + accumulators[resourceType] = acc + } + + acc.totalRequests++ + + if respTime, ok := event.EventData["response_time"].(float64); ok { + acc.totalResponseTime += respTime + } + + if event.EventType == "error" || event.EventType == tenant.TenantEventQuotaExceeded { + acc.errorCount++ + } + } + + // Convert accumulators to ResourcePerformance + for resourceType, acc := range accumulators { + avgResponseTime := 0.0 + if acc.totalRequests > 0 { + avgResponseTime = acc.totalResponseTime / float64(acc.totalRequests) + } + + errorRate := 0.0 + if acc.totalRequests > 0 { + errorRate = (float64(acc.errorCount) / float64(acc.totalRequests)) * 100 + } + + resourcePerformance[resourceType] = &tenant.ResourcePerformance{ + ResourceType: resourceType, + ResponseTime: avgResponseTime, + Throughput: float64(acc.totalRequests), + ErrorRate: errorRate, + } } return resourcePerformance diff --git a/apps/carrier-connector/internal/smdp/load_balancer.go b/apps/carrier-connector/internal/smdp/load_balancer.go index 6af6b6a..988b91f 100644 --- a/apps/carrier-connector/internal/smdp/load_balancer.go +++ b/apps/carrier-connector/internal/smdp/load_balancer.go @@ -8,8 +8,9 @@ import ( // LoadBalancer implements different load balancing strategies for carrier selection type LoadBalancer struct { - strategy LoadBalancingStrategy - rand *rand.Rand + strategy LoadBalancingStrategy + rand *rand.Rand + rrCounter uint64 // round-robin request counter } // LoadBalancingStrategy defines the load balancing algorithm @@ -55,9 +56,9 @@ func (lb *LoadBalancer) SelectCarrier(carriers []*Carrier, req *ProfileRequest) // 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 + idx := lb.rrCounter % uint64(len(carriers)) + lb.rrCounter++ + return carriers[idx] } // weightedRoundRobinSelect selects carrier based on priority and health