feat: Add pricing repository, tenant-aware transaction support, and pricing integration service

- Add GormPricingRepository with CreateRule, GetRule, UpdateRule, DeleteRule, ListRules, CountRules methods
- Add GetActiveRules, GetRulesByType methods for retrieving active pricing rules by tenant and type
- Add TenantScopedDB wrapper with automatic tenant filtering for tenant-aware database operations
- Implement TenantScopedTransaction with tenant context propagation and automatic filtering

apps/carrier-connector/internal/repository/pricing_repository.go

@@ -0,0 +1,124 @@
+package repository
+
+import (
+	"context"
+	"time"
+
+	"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/pricing"
+	"gorm.io/gorm"
+)
+
+// GormPricingRepository implements the pricing repository interface using GORM
+type GormPricingRepository struct {
+	db *gorm.DB
+}
+
+// NewGormPricingRepository creates a new GORM pricing repository
+func NewGormPricingRepository(db *gorm.DB) pricing.Repository {
+	return &GormPricingRepository{db: db}
+}
+
+// CreateRule creates a new pricing rule
+func (r *GormPricingRepository) CreateRule(ctx context.Context, rule *pricing.PricingRule) error {
+	rule.CreatedAt = time.Now()
+	rule.UpdatedAt = time.Now()
+	return r.db.WithContext(ctx).Create(rule).Error
+}
+
+// GetRule retrieves a pricing rule by ID
+func (r *GormPricingRepository) GetRule(ctx context.Context, id string) (*pricing.PricingRule, error) {
+	var rule pricing.PricingRule
+	err := r.db.WithContext(ctx).Where("id = ?", id).First(&rule).Error
+	if err != nil {
+		return nil, err
+	}
+	return &rule, nil
+}
+
+// UpdateRule updates an existing pricing rule
+func (r *GormPricingRepository) UpdateRule(ctx context.Context, rule *pricing.PricingRule) error {
+	rule.UpdatedAt = time.Now()
+	return r.db.WithContext(ctx).Save(rule).Error
+}
+
+// DeleteRule deletes a pricing rule
+func (r *GormPricingRepository) DeleteRule(ctx context.Context, id string) error {
+	return r.db.WithContext(ctx).Delete(&pricing.PricingRule{}, "id = ?", id).Error
+}
+
+// ListRules lists pricing rules with filtering
+func (r *GormPricingRepository) ListRules(ctx context.Context, filter *pricing.PricingFilter) ([]*pricing.PricingRule, error) {
+	query := r.db.WithContext(ctx).Model(&pricing.PricingRule{})
+
+	// Apply filters
+	if filter.TenantID != "" {
+		query = query.Where("tenant_id = ?", filter.TenantID)
+	}
+	if filter.Type != "" {
+		query = query.Where("type = ?", filter.Type)
+	}
+	if filter.IsActive != nil {
+		query = query.Where("is_active = ?", *filter.IsActive)
+	}
+	if filter.Priority != nil {
+		query = query.Where("priority = ?", *filter.Priority)
+	}
+
+	// Apply ordering
+	query = query.Order("priority DESC, created_at DESC")
+
+	// Apply pagination
+	if filter.Limit > 0 {
+		query = query.Limit(filter.Limit)
+	}
+	if filter.Offset > 0 {
+		query = query.Offset(filter.Offset)
+	}
+
+	var rules []*pricing.PricingRule
+	err := query.Find(&rules).Error
+	return rules, err
+}
+
+// CountRules counts pricing rules with filtering
+func (r *GormPricingRepository) CountRules(ctx context.Context, filter *pricing.PricingFilter) (int, error) {
+	query := r.db.WithContext(ctx).Model(&pricing.PricingRule{})
+
+	// Apply filters
+	if filter.TenantID != "" {
+		query = query.Where("tenant_id = ?", filter.TenantID)
+	}
+	if filter.Type != "" {
+		query = query.Where("type = ?", filter.Type)
+	}
+	if filter.IsActive != nil {
+		query = query.Where("is_active = ?", *filter.IsActive)
+	}
+	if filter.Priority != nil {
+		query = query.Where("priority = ?", *filter.Priority)
+	}
+
+	var count int64
+	err := query.Count(&count).Error
+	return int(count), err
+}
+
+// GetActiveRules retrieves all active rules for a tenant
+func (r *GormPricingRepository) GetActiveRules(ctx context.Context, tenantID string) ([]*pricing.PricingRule, error) {
+	var rules []*pricing.PricingRule
+	err := r.db.WithContext(ctx).
+		Where("tenant_id = ? AND is_active = ?", tenantID, true).
+		Order("priority DESC, created_at DESC").
+		Find(&rules).Error
+	return rules, err
+}
+
+// GetRulesByType retrieves rules by type for a tenant
+func (r *GormPricingRepository) GetRulesByType(ctx context.Context, tenantID string, ruleType pricing.RuleType) ([]*pricing.PricingRule, error) {
+	var rules []*pricing.PricingRule
+	err := r.db.WithContext(ctx).
+		Where("tenant_id = ? AND type = ? AND is_active = ?", tenantID, ruleType, true).
+		Order("priority DESC, created_at DESC").
+		Find(&rules).Error
+	return rules, err
+}

apps/carrier-connector/internal/repository/tenant_aware_repository.go

@@ -67,13 +67,54 @@ func (r *TenantAwareRepository) TenantScopedTransaction(ctx context.Context, fn
 	return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
 		// Apply tenant filtering to all operations within the transaction
 		if r.tenantID != "" {
-			// This would need to be implemented based on specific requirements
-			// For now, we'll rely on the tenant-scoped queries within the transaction
+			// Set tenant context for the transaction
+			tx = tx.Set("tenant_id", r.tenantID)
+
+			// Create a wrapped DB instance that automatically applies tenant filtering
+			wrappedTx := &TenantScopedDB{
+				DB:       tx,
+				tenantID: r.tenantID,
+			}
+
+			// Use the wrapped transaction for all operations
+			// Since the function expects *gorm.DB, we need to embed the wrapper properly
+			return fn(wrappedTx.DB.Scopes(func(db *gorm.DB) *gorm.DB {
+				// Apply tenant filtering to all queries
+				return db.Where("tenant_id = ?", r.tenantID)
+			}))
 		}
 		return fn(tx)
 	})
 }
 
+// TenantScopedDB wraps GORM DB with automatic tenant filtering
+type TenantScopedDB struct {
+	*gorm.DB
+	tenantID string
+}
+
+// Table overrides the Table method to add tenant filtering
+func (t *TenantScopedDB) Table(name string) *gorm.DB {
+	// Add tenant filtering for tables that have tenant_id column
+	tenantFilteredTables := map[string]bool{
+		"tenants":                 true,
+		"tenant_users":            true,
+		"tenant_configs":          true,
+		"tenant_events":           true,
+		"profiles":                true,
+		"rate_plans":              true,
+		"rate_plan_subscriptions": true,
+		"carriers":                true,
+		"pricing_rules":           true,
+	}
+
+	if tenantFilteredTables[name] {
+		return t.DB.Table(name).Where("tenant_id = ?", t.tenantID)
+	}
+
+	return t.DB.Table(name)
+}
+
 // ValidateResourceAccess validates that a resource belongs to a tenant
 func (v *TenantResourceValidator) ValidateResourceAccess(ctx context.Context, tenantID, resourceType, resourceID string) error {
 	switch resourceType {
@@ -91,25 +132,93 @@ func (v *TenantResourceValidator) ValidateResourceAccess(ctx context.Context, te
 }
 
 func (v *TenantResourceValidator) validateProfileAccess(ctx context.Context, tenantID, profileID string) error {
-	// This would check if the profile belongs to the tenant
-	// Implementation depends on the actual profile schema
+	if profileID == "" {
+		return fmt.Errorf("profile ID cannot be empty")
+	}
+
+	// Check if profile exists and belongs to tenant
+	var count int64
+	err := v.db.WithContext(ctx).
+		Table("profiles").
+		Where("iccid = ? AND tenant_id = ?", profileID, tenantID).
+		Count(&count).Error
+
+	if err != nil {
+		return fmt.Errorf("failed to validate profile access: %w", err)
+	}
+
+	if count == 0 {
+		return fmt.Errorf("profile %s not found or does not belong to tenant %s", profileID, tenantID)
+	}
+
 	return nil
 }
 
 func (v *TenantResourceValidator) validateRatePlanAccess(ctx context.Context, tenantID, ratePlanID string) error {
-	// This would check if the rate plan belongs to the tenant
-	// Implementation depends on the actual rate plan schema
+	if ratePlanID == "" {
+		return fmt.Errorf("rate plan ID cannot be empty")
+	}
+
+	// Check if rate plan exists and belongs to tenant
+	var count int64
+	err := v.db.WithContext(ctx).
+		Table("rate_plans").
+		Where("id = ? AND tenant_id = ?", ratePlanID, tenantID).
+		Count(&count).Error
+
+	if err != nil {
+		return fmt.Errorf("failed to validate rate plan access: %w", err)
+	}
+
+	if count == 0 {
+		return fmt.Errorf("rate plan %s not found or does not belong to tenant %s", ratePlanID, tenantID)
+	}
+
 	return nil
 }
 
 func (v *TenantResourceValidator) validateCarrierAccess(ctx context.Context, tenantID, carrierID string) error {
-	// This would check if the carrier belongs to the tenant
-	// Implementation depends on the actual carrier schema
+	if carrierID == "" {
+		return fmt.Errorf("carrier ID cannot be empty")
+	}
+
+	// Check if carrier exists and belongs to tenant
+	var count int64
+	err := v.db.WithContext(ctx).
+		Table("carriers").
+		Where("id = ? AND tenant_id = ?", carrierID, tenantID).
+		Count(&count).Error
+
+	if err != nil {
+		return fmt.Errorf("failed to validate carrier access: %w", err)
+	}
+
+	if count == 0 {
+		return fmt.Errorf("carrier %s not found or does not belong to tenant %s", carrierID, tenantID)
+	}
+
 	return nil
 }
 
 func (v *TenantResourceValidator) validateSubscriptionAccess(ctx context.Context, tenantID, subscriptionID string) error {
-	// This would check if the subscription belongs to the tenant
-	// Implementation depends on the actual subscription schema
+	if subscriptionID == "" {
+		return fmt.Errorf("subscription ID cannot be empty")
+	}
+
+	// Check if subscription exists and belongs to tenant
+	var count int64
+	err := v.db.WithContext(ctx).
+		Table("rate_plan_subscriptions").
+		Where("id = ? AND profile_id IN (SELECT iccid FROM profiles WHERE tenant_id = ?)", subscriptionID, tenantID).
+		Count(&count).Error
+
+	if err != nil {
+		return fmt.Errorf("failed to validate subscription access: %w", err)
+	}
+
+	if count == 0 {
+		return fmt.Errorf("subscription %s not found or does not belong to tenant %s", subscriptionID, tenantID)
+	}
+
 	return nil
 }

apps/carrier-connector/internal/services/pricing_integration.go

@@ -0,0 +1,212 @@
+package services
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/pricing"
+	"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/rateplan"
+	"github.com/sirupsen/logrus"
+)
+
+// PricingIntegration integrates dynamic pricing with rate plans
+type PricingIntegration struct {
+	pricingService  pricing.Service
+	rateplanService rateplan.Service
+	logger          *logrus.Logger
+}
+
+// NewPricingIntegration creates a new pricing integration service
+func NewPricingIntegration(
+	pricingService pricing.Service,
+	rateplanService rateplan.Service,
+	logger *logrus.Logger,
+) *PricingIntegration {
+	return &PricingIntegration{
+		pricingService:  pricingService,
+		rateplanService: rateplanService,
+		logger:          logger,
+	}
+}
+
+// CalculateRatePlanPrice calculates the price for a rate plan with dynamic pricing
+func (pi *PricingIntegration) CalculateRatePlanPrice(ctx context.Context, tenantID, ratePlanID string, quantity int, customerID string) (*pricing.PricingResult, error) {
+	// Get rate plan
+	ratePlan, err := pi.rateplanService.GetRatePlan(ctx, ratePlanID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get rate plan: %w", err)
+	}
+
+	// Create pricing context
+	pricingCtx := &pricing.PricingContext{
+		TenantID:   tenantID,
+		CustomerID: customerID,
+		ProductID:  ratePlanID,
+		BasePrice:  ratePlan.BasePrice,
+		Currency:   ratePlan.Currency,
+		Quantity:   quantity,
+		Location:   ratePlan.Region,
+		Time:       getCurrentTime(),
+		Metadata: map[string]any{
+			"rate_plan_name": ratePlan.Name,
+			"carrier_id":     ratePlan.CarrierID,
+			"plan_type":      ratePlan.PlanType,
+		},
+	}
+
+	// Calculate price with dynamic pricing
+	result, err := pi.pricingService.CalculatePrice(ctx, pricingCtx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to calculate price: %w", err)
+	}
+
+	pi.logger.WithFields(logrus.Fields{
+		"tenant_id":      tenantID,
+		"rate_plan_id":   ratePlanID,
+		"customer_id":    customerID,
+		"original_price": result.OriginalPrice,
+		"final_price":    result.FinalPrice,
+		"discount":       result.Discount,
+	}).Info("Rate plan price calculated with dynamic pricing")
+
+	return result, nil
+}
+
+// ApplyPricingToSubscription applies pricing rules to a subscription
+func (pi *PricingIntegration) ApplyPricingToSubscription(ctx context.Context, subscription *rateplan.RatePlanSubscription) (*rateplan.RatePlanSubscription, error) {
+	// Get rate plan
+	_, err := pi.rateplanService.GetRatePlan(ctx, subscription.RatePlanID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get rate plan: %w", err)
+	}
+
+	// Calculate pricing
+	result, err := pi.CalculateRatePlanPrice(ctx, subscription.ProfileID, subscription.RatePlanID, 1, subscription.ProfileID)
+	if err != nil {
+		return nil, err
+	}
+
+	// Update subscription metadata with pricing information
+	if subscription.Metadata == nil {
+		subscription.Metadata = make(map[string]any)
+	}
+
+	subscription.Metadata["dynamic_pricing"] = map[string]any{
+		"original_price": result.OriginalPrice,
+		"final_price":    result.FinalPrice,
+		"discount":       result.Discount,
+		"applied_rules":  result.AppliedRules,
+		"calculated_at":  getCurrentTime(),
+	}
+
+	return subscription, nil
+}
+
+// CreatePricingRuleFromRatePlan creates a pricing rule based on a rate plan
+func (pi *PricingIntegration) CreatePricingRuleFromRatePlan(ctx context.Context, tenantID string, ratePlanID string, ruleType pricing.RuleType, conditions pricing.RuleConditions, actions pricing.RuleActions) (*pricing.PricingRule, error) {
+	// Get rate plan
+	ratePlan, err := pi.rateplanService.GetRatePlan(ctx, ratePlanID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get rate plan: %w", err)
+	}
+
+	// Create pricing rule
+	rule := &pricing.PricingRule{
+		ID:          generateRuleID(),
+		Name:        fmt.Sprintf("Dynamic pricing for %s", ratePlan.Name),
+		Description: fmt.Sprintf("Automatically generated pricing rule for rate plan %s", ratePlan.Name),
+		TenantID:    tenantID,
+		Type:        ruleType,
+		Priority:    100, // High priority for auto-generated rules
+		IsActive:    true,
+		Conditions:  conditions,
+		Actions:     actions,
+		Metadata: map[string]any{
+			"rate_plan_id":   ratePlanID,
+			"rate_plan_name": ratePlan.Name,
+			"auto_generated": true,
+			"generated_at":   getCurrentTime(),
+		},
+	}
+
+	// Create rule
+	createdRule, err := pi.pricingService.CreateRule(ctx, rule)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create pricing rule: %w", err)
+	}
+
+	pi.logger.WithFields(logrus.Fields{
+		"tenant_id":    tenantID,
+		"rate_plan_id": ratePlanID,
+		"rule_id":      createdRule.ID,
+		"rule_type":    ruleType,
+	}).Info("Pricing rule created from rate plan")
+
+	return createdRule, nil
+}
+
+// GetPricingEffectiveness analyzes the effectiveness of pricing rules
+func (pi *PricingIntegration) GetPricingEffectiveness(ctx context.Context, tenantID string) (*PricingEffectiveness, error) {
+	// Get pricing analytics
+	analytics, err := pi.pricingService.GetAnalytics(ctx, tenantID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get pricing analytics: %w", err)
+	}
+
+	// Get rate plan analytics (placeholder - would need actual implementation)
+	ratePlanAnalytics := &RatePlanPricingAnalytics{
+		TotalRatePlans:   0,
+		PlansWithPricing: 0,
+		AverageDiscount:  0.0,
+		TotalSavings:     0.0,
+		ConversionRate:   0.0,
+	}
+
+	// Calculate effectiveness
+	effectiveness := &PricingEffectiveness{
+		TotalRules:            analytics.TotalRules,
+		ActiveRules:           analytics.ActiveRules,
+		TotalRatePlans:        ratePlanAnalytics.TotalRatePlans,
+		PlansWithPricing:      ratePlanAnalytics.PlansWithPricing,
+		AverageDiscountRate:   analytics.DiscountStats.AverageDiscount,
+		TotalSavings:          analytics.DiscountStats.TotalDiscountValue,
+		RulesByType:           analytics.RulesByType,
+		ConversionImprovement: ratePlanAnalytics.ConversionRate,
+		GeneratedAt:           getCurrentTime(),
+	}
+
+	return effectiveness, nil
+}
+
+// Supporting types
+
+type PricingEffectiveness struct {
+	TotalRules            int            `json:"total_rules"`
+	ActiveRules           int            `json:"active_rules"`
+	TotalRatePlans        int            `json:"total_rate_plans"`
+	PlansWithPricing      int            `json:"plans_with_pricing"`
+	AverageDiscountRate   float64        `json:"average_discount_rate"`
+	TotalSavings          float64        `json:"total_savings"`
+	RulesByType           map[string]int `json:"rules_by_type"`
+	ConversionImprovement float64        `json:"conversion_improvement"`
+	GeneratedAt           interface{}    `json:"generated_at"`
+}
+
+type RatePlanPricingAnalytics struct {
+	TotalRatePlans   int     `json:"total_rate_plans"`
+	PlansWithPricing int     `json:"plans_with_pricing"`
+	AverageDiscount  float64 `json:"average_discount"`
+	TotalSavings     float64 `json:"total_savings"`
+	ConversionRate   float64 `json:"conversion_rate"`
+}
+
+// Helper functions
+
+func generateRuleID() string {
+	return fmt.Sprintf("rule_%d", getCurrentTime().UnixNano())
+}
+
+func getCurrentTime() time.Time {
+	return time.Now()
+}

apps/carrier-connector/internal/services/pricing_rule_engine.go

@@ -0,0 +1,212 @@
+package services
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/pricing"
+)
+
+// PricingRuleEngine implements the rule evaluation logic
+type PricingRuleEngine struct {
+	validator pricing.PricingValidator
+}
+
+// NewPricingRuleEngine creates a new pricing rule engine
+func NewPricingRuleEngine(validator pricing.PricingValidator) pricing.RuleEngine {
+	return &PricingRuleEngine{
+		validator: validator,
+	}
+}
+
+// EvaluateRule checks if a rule should be applied to the given context
+func (e *PricingRuleEngine) EvaluateRule(ctx context.Context, rule *pricing.PricingRule, pricingCtx *pricing.PricingContext) (bool, error) {
+	if !rule.IsActive {
+		return false, nil
+	}
+
+	// Validate conditions
+	matches, err := e.ValidateConditions(ctx, rule.Conditions, pricingCtx)
+	if err != nil {
+		return false, fmt.Errorf("failed to validate conditions: %w", err)
+	}
+
+	return matches, nil
+}
+
+// ApplyRule applies a rule to adjust the price
+func (e *PricingRuleEngine) ApplyRule(ctx context.Context, rule *pricing.PricingRule, pricingCtx *pricing.PricingContext, currentPrice float64) (float64, error) {
+	adjustedPrice, err := e.ExecuteActions(ctx, rule.Actions, currentPrice)
+	if err != nil {
+		return currentPrice, fmt.Errorf("failed to execute actions: %w", err)
+	}
+
+	return adjustedPrice, nil
+}
+
+// ValidateConditions checks if the conditions match the pricing context
+func (e *PricingRuleEngine) ValidateConditions(ctx context.Context, conditions pricing.RuleConditions, pricingCtx *pricing.PricingContext) (bool, error) {
+	// Check time range
+	if conditions.TimeRange != nil {
+		if !e.isWithinTimeRange(conditions.TimeRange, pricingCtx.Time) {
+			return false, nil
+		}
+	}
+
+	// Check geography
+	if len(conditions.Geography) > 0 {
+		if !e.isInGeography(conditions.Geography, pricingCtx.Location) {
+			return false, nil
+		}
+	}
+
+	// Check customer type
+	if len(conditions.CustomerType) > 0 {
+		customerType, ok := pricingCtx.Metadata["customer_type"].(string)
+		if !ok || !e.isInCustomerType(conditions.CustomerType, customerType) {
+			return false, nil
+		}
+	}
+
+	// Check volume
+	if conditions.Volume != nil {
+		if !e.isWithinVolumeRange(conditions.Volume, pricingCtx.Quantity) {
+			return false, nil
+		}
+	}
+
+	// Check usage pattern
+	if conditions.UsagePattern != nil {
+		if !e.matchesUsagePattern(conditions.UsagePattern, pricingCtx.Time) {
+			return false, nil
+		}
+	}
+
+	return true, nil
+}
+
+// ExecuteActions applies the pricing adjustments
+func (e *PricingRuleEngine) ExecuteActions(ctx context.Context, actions pricing.RuleActions, currentPrice float64) (float64, error) {
+	switch actions.AdjustmentType {
+	case pricing.AdjustmentTypePercentage:
+		adjustedPrice := currentPrice * (1 - actions.Value/100)
+		if actions.Limit != nil && adjustedPrice < *actions.Limit {
+			adjustedPrice = *actions.Limit
+		}
+		return adjustedPrice, nil
+
+	case pricing.AdjustmentTypeFixed:
+		adjustedPrice := currentPrice - actions.Value
+		if actions.Limit != nil && adjustedPrice < *actions.Limit {
+			adjustedPrice = *actions.Limit
+		}
+		return adjustedPrice, nil
+
+	case pricing.AdjustmentTypeMultiply:
+		adjustedPrice := currentPrice * actions.Value
+		if actions.Limit != nil && adjustedPrice > *actions.Limit {
+			adjustedPrice = *actions.Limit
+		}
+		return adjustedPrice, nil
+
+	case pricing.AdjustmentTypeOverride:
+		if actions.NewPrice != nil {
+			return *actions.NewPrice, nil
+		}
+		return currentPrice, fmt.Errorf("override action requires new_price")
+
+	default:
+		return currentPrice, fmt.Errorf("unknown adjustment type: %s", actions.AdjustmentType)
+	}
+}
+
+// Helper methods for condition validation
+
+func (e *PricingRuleEngine) isWithinTimeRange(timeRange *pricing.TimeRange, checkTime time.Time) bool {
+	if checkTime.Before(timeRange.Start) || checkTime.After(timeRange.End) {
+		return false
+	}
+
+	if len(timeRange.Days) > 0 {
+		day := strings.ToLower(checkTime.Weekday().String())
+		for _, allowedDay := range timeRange.Days {
+			if strings.ToLower(allowedDay) == day {
+				return true
+			}
+		}
+		return false
+	}
+
+	return true
+}
+
+func (e *PricingRuleEngine) isInGeography(geography []string, location string) bool {
+	if location == "" {
+		return false
+	}
+
+	for _, geo := range geography {
+		if strings.EqualFold(geo, location) {
+			return true
+		}
+	}
+	return false
+}
+
+func (e *PricingRuleEngine) isInCustomerType(customerTypes []string, customerType string) bool {
+	for _, ct := range customerTypes {
+		if strings.EqualFold(ct, customerType) {
+			return true
+		}
+	}
+	return false
+}
+
+func (e *PricingRuleEngine) isWithinVolumeRange(volumeRange *pricing.VolumeRange, quantity int) bool {
+	return quantity >= volumeRange.Min && quantity <= volumeRange.Max
+}
+
+func (e *PricingRuleEngine) matchesUsagePattern(pattern *pricing.UsagePattern, checkTime time.Time) bool {
+	hour := checkTime.Hour()
+	
+	// Check peak hours
+	if len(pattern.PeakHours) > 0 {
+		for _, peakHour := range pattern.PeakHours {
+			if e.isHourInRange(hour, peakHour) {
+				return true
+			}
+		}
+	}
+
+	// Check off-peak hours
+	if len(pattern.OffPeakHours) > 0 {
+		for _, offPeakHour := range pattern.OffPeakHours {
+			if e.isHourInRange(hour, offPeakHour) {
+				return true
+			}
+		}
+	}
+
+	return false
+}
+
+func (e *PricingRuleEngine) isHourInRange(hour int, hourRange string) bool {
+	// Parse hour range like "9-17", "18-23", etc.
+	parts := strings.Split(hourRange, "-")
+	if len(parts) != 2 {
+		return false
+	}
+
+	start := e.parseHour(parts[0])
+	end := e.parseHour(parts[1])
+	
+	return hour >= start && hour <= end
+}
+
+func (e *PricingRuleEngine) parseHour(hourStr string) int {
+	hour := 0
+	fmt.Sscanf(hourStr, "%d", &hour)
+	return hour
+}

apps/carrier-connector/internal/services/pricing_service.go

@@ -0,0 +1,262 @@
+package services
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/pricing"
+	"github.com/sirupsen/logrus"
+)
+
+// PricingService implements the pricing business logic
+type PricingService struct {
+	repository pricing.Repository
+	engine     pricing.RuleEngine
+	validator  pricing.PricingValidator
+	logger     *logrus.Logger
+}
+
+// NewPricingService creates a new pricing service
+func NewPricingService(
+	repository pricing.Repository,
+	engine pricing.RuleEngine,
+	validator pricing.PricingValidator,
+	logger *logrus.Logger,
+) pricing.Service {
+	return &PricingService{
+		repository: repository,
+		engine:     engine,
+		validator:  validator,
+		logger:     logger,
+	}
+}
+
+// CreateRule creates a new pricing rule
+func (s *PricingService) CreateRule(ctx context.Context, rule *pricing.PricingRule) (*pricing.PricingRule, error) {
+	// Validate rule
+	if err := s.validator.ValidateRule(ctx, rule); err != nil {
+		return nil, fmt.Errorf("validation failed: %w", err)
+	}
+
+	// Create rule
+	if err := s.repository.CreateRule(ctx, rule); err != nil {
+		s.logger.WithError(err).Error("Failed to create pricing rule")
+		return nil, fmt.Errorf("failed to create rule: %w", err)
+	}
+
+	s.logger.WithFields(logrus.Fields{
+		"rule_id":   rule.ID,
+		"tenant_id": rule.TenantID,
+		"type":      rule.Type,
+	}).Info("Pricing rule created successfully")
+
+	return rule, nil
+}
+
+// GetRule retrieves a pricing rule by ID
+func (s *PricingService) GetRule(ctx context.Context, id string) (*pricing.PricingRule, error) {
+	rule, err := s.repository.GetRule(ctx, id)
+	if err != nil {
+		s.logger.WithError(err).WithField("rule_id", id).Error("Failed to get pricing rule")
+		return nil, err
+	}
+
+	return rule, nil
+}
+
+// UpdateRule updates an existing pricing rule
+func (s *PricingService) UpdateRule(ctx context.Context, id string, rule *pricing.PricingRule) (*pricing.PricingRule, error) {
+	// Validate rule
+	if err := s.validator.ValidateRule(ctx, rule); err != nil {
+		return nil, fmt.Errorf("validation failed: %w", err)
+	}
+
+	// Update rule
+	if err := s.repository.UpdateRule(ctx, rule); err != nil {
+		s.logger.WithError(err).WithField("rule_id", id).Error("Failed to update pricing rule")
+		return nil, fmt.Errorf("failed to update rule: %w", err)
+	}
+
+	s.logger.WithFields(logrus.Fields{
+		"rule_id":   rule.ID,
+		"tenant_id": rule.TenantID,
+		"type":      rule.Type,
+	}).Info("Pricing rule updated successfully")
+
+	return rule, nil
+}
+
+// DeleteRule deletes a pricing rule
+func (s *PricingService) DeleteRule(ctx context.Context, id string) error {
+	// Get rule for logging
+	rule, err := s.repository.GetRule(ctx, id)
+	if err != nil {
+		return err
+	}
+
+	// Delete rule
+	if err := s.repository.DeleteRule(ctx, id); err != nil {
+		s.logger.WithError(err).WithField("rule_id", id).Error("Failed to delete pricing rule")
+		return fmt.Errorf("failed to delete rule: %w", err)
+	}
+
+	s.logger.WithFields(logrus.Fields{
+		"rule_id":   rule.ID,
+		"tenant_id": rule.TenantID,
+		"type":      rule.Type,
+	}).Info("Pricing rule deleted successfully")
+
+	return nil
+}
+
+// ListRules lists pricing rules with filtering
+func (s *PricingService) ListRules(ctx context.Context, filter *pricing.PricingFilter) ([]*pricing.PricingRule, error) {
+	rules, err := s.repository.ListRules(ctx, filter)
+	if err != nil {
+		s.logger.WithError(err).Error("Failed to list pricing rules")
+		return nil, fmt.Errorf("failed to list rules: %w", err)
+	}
+
+	return rules, nil
+}
+
+// CalculatePrice calculates the final price based on active rules
+func (s *PricingService) CalculatePrice(ctx context.Context, pricingCtx *pricing.PricingContext) (*pricing.PricingResult, error) {
+	// Validate context
+	if err := s.validator.ValidateContext(ctx, pricingCtx); err != nil {
+		return nil, fmt.Errorf("invalid pricing context: %w", err)
+	}
+
+	// Get active rules for tenant
+	rules, err := s.repository.GetActiveRules(ctx, pricingCtx.TenantID)
+	if err != nil {
+		s.logger.WithError(err).WithField("tenant_id", pricingCtx.TenantID).Error("Failed to get active rules")
+		return nil, fmt.Errorf("failed to get active rules: %w", err)
+	}
+
+	// Apply rules to calculate final price
+	result, err := s.ApplyRules(ctx, pricingCtx, rules)
+	if err != nil {
+		return nil, err
+	}
+
+	s.logger.WithFields(logrus.Fields{
+		"tenant_id":      pricingCtx.TenantID,
+		"product_id":     pricingCtx.ProductID,
+		"original_price": result.OriginalPrice,
+		"final_price":    result.FinalPrice,
+		"rules_applied":  len(result.AppliedRules),
+	}).Info("Price calculated successfully")
+
+	return result, nil
+}
+
+// ApplyRules applies specific rules to a pricing context
+func (s *PricingService) ApplyRules(ctx context.Context, pricingCtx *pricing.PricingContext, rules []*pricing.PricingRule) (*pricing.PricingResult, error) {
+	result := &pricing.PricingResult{
+		OriginalPrice: pricingCtx.BasePrice,
+		AdjustedPrice: pricingCtx.BasePrice,
+		FinalPrice:    pricingCtx.BasePrice,
+		Currency:      pricingCtx.Currency,
+		AppliedRules:  make([]pricing.AppliedRule, 0),
+		Metadata:      make(map[string]any),
+	}
+
+	currentPrice := pricingCtx.BasePrice
+
+	// Apply rules in priority order
+	for _, rule := range rules {
+		shouldApply, err := s.engine.EvaluateRule(ctx, rule, pricingCtx)
+		if err != nil {
+			s.logger.WithError(err).WithField("rule_id", rule.ID).Error("Failed to evaluate rule")
+			continue
+		}
+
+		if shouldApply {
+			adjustedPrice, err := s.engine.ApplyRule(ctx, rule, pricingCtx, currentPrice)
+			if err != nil {
+				s.logger.WithError(err).WithField("rule_id", rule.ID).Error("Failed to apply rule")
+				continue
+			}
+
+			// Calculate discount amount
+			discount := currentPrice - adjustedPrice
+
+			// Update result
+			currentPrice = adjustedPrice
+			result.AppliedRules = append(result.AppliedRules, pricing.AppliedRule{
+				RuleID:     rule.ID,
+				RuleName:   rule.Name,
+				Type:       string(rule.Type),
+				Adjustment: discount,
+			})
+
+			s.logger.WithFields(logrus.Fields{
+				"rule_id":    rule.ID,
+				"rule_name":  rule.Name,
+				"adjustment": discount,
+				"new_price":  adjustedPrice,
+			}).Debug("Rule applied")
+		}
+	}
+
+	// Finalize result
+	result.FinalPrice = currentPrice
+	result.Discount = result.OriginalPrice - result.FinalPrice
+
+	return result, nil
+}
+
+// ValidateRule validates a pricing rule
+func (s *PricingService) ValidateRule(ctx context.Context, rule *pricing.PricingRule) error {
+	return s.validator.ValidateRule(ctx, rule)
+}
+
+// GetAnalytics retrieves pricing analytics for a tenant
+func (s *PricingService) GetAnalytics(ctx context.Context, tenantID string) (*pricing.PricingAnalytics, error) {
+	// Get all rules for tenant
+	allRules, err := s.repository.ListRules(ctx, &pricing.PricingFilter{
+		TenantID: tenantID,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to get rules for analytics: %w", err)
+	}
+
+	// Get active rules
+	activeRules, err := s.repository.GetActiveRules(ctx, tenantID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get active rules for analytics: %w", err)
+	}
+
+	// Calculate analytics
+	analytics := &pricing.PricingAnalytics{
+		TotalRules:  len(allRules),
+		ActiveRules: len(activeRules),
+		RulesByType: make(map[string]int),
+		UsageByRule: make(map[string]int64),
+		GeneratedAt: s.getCurrentTime(),
+	}
+
+	// Count rules by type
+	for _, rule := range allRules {
+		ruleType := string(rule.Type)
+		analytics.RulesByType[ruleType]++
+	}
+
+	// TODO: Calculate actual usage statistics from pricing history
+	// For now, return placeholder data
+	analytics.DiscountStats = pricing.DiscountStatistics{
+		TotalDiscounts:     0,
+		AverageDiscount:    0.0,
+		LargestDiscount:    0.0,
+		SmallestDiscount:   0.0,
+		TotalDiscountValue: 0.0,
+	}
+
+	return analytics, nil
+}
+
+func (s *PricingService) getCurrentTime() time.Time {
+	return time.Now()
+}

apps/carrier-connector/internal/services/pricing_validator.go

@@ -0,0 +1,243 @@
+package services
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/pricing"
+)
+
+// PricingValidator implements validation for pricing rules and contexts
+type PricingValidator struct{}
+
+// NewPricingValidator creates a new pricing validator
+func NewPricingValidator() pricing.PricingValidator {
+	return &PricingValidator{}
+}
+
+// ValidateRule validates a pricing rule
+func (v *PricingValidator) ValidateRule(ctx context.Context, rule *pricing.PricingRule) error {
+	if rule.ID == "" {
+		return errors.New("rule ID is required")
+	}
+
+	if rule.Name == "" {
+		return errors.New("rule name is required")
+	}
+
+	if rule.TenantID == "" {
+		return errors.New("tenant ID is required")
+	}
+
+	if rule.Type == "" {
+		return errors.New("rule type is required")
+	}
+
+	// Validate rule type
+	validTypes := map[pricing.RuleType]bool{
+		pricing.RuleTypePercentageDiscount: true,
+		pricing.RuleTypeFixedDiscount:      true,
+		pricing.RuleTypeMultiplier:         true,
+		pricing.RuleTypeTieredPricing:      true,
+		pricing.RuleTypeDynamicPricing:     true,
+		pricing.RuleTypeConditionalPricing: true,
+	}
+
+	if !validTypes[rule.Type] {
+		return fmt.Errorf("invalid rule type: %s", rule.Type)
+	}
+
+	// Validate conditions
+	if err := v.ValidateConditions(ctx, rule.Conditions); err != nil {
+		return fmt.Errorf("invalid conditions: %w", err)
+	}
+
+	// Validate actions
+	if err := v.ValidateActions(ctx, rule.Actions); err != nil {
+		return fmt.Errorf("invalid actions: %w", err)
+	}
+
+	return nil
+}
+
+// ValidateContext validates a pricing context
+func (v *PricingValidator) ValidateContext(ctx context.Context, pricingCtx *pricing.PricingContext) error {
+	if pricingCtx.TenantID == "" {
+		return errors.New("tenant ID is required")
+	}
+
+	if pricingCtx.ProductID == "" {
+		return errors.New("product ID is required")
+	}
+
+	if pricingCtx.BasePrice < 0 {
+		return errors.New("base price cannot be negative")
+	}
+
+	if pricingCtx.Currency == "" {
+		return errors.New("currency is required")
+	}
+
+	if pricingCtx.Quantity <= 0 {
+		return errors.New("quantity must be positive")
+	}
+
+	if pricingCtx.Time.IsZero() {
+		return errors.New("time is required")
+	}
+
+	return nil
+}
+
+// ValidateConditions validates rule conditions
+func (v *PricingValidator) ValidateConditions(ctx context.Context, conditions pricing.RuleConditions) error {
+	// Validate time range
+	if conditions.TimeRange != nil {
+		if conditions.TimeRange.Start.IsZero() || conditions.TimeRange.End.IsZero() {
+			return errors.New("time range requires both start and end times")
+		}
+
+		if conditions.TimeRange.Start.After(conditions.TimeRange.End) {
+			return errors.New("time range start must be before end")
+		}
+
+		// Validate days
+		validDays := map[string]bool{
+			"monday":    true,
+			"tuesday":   true,
+			"wednesday": true,
+			"thursday":  true,
+			"friday":    true,
+			"saturday":  true,
+			"sunday":    true,
+		}
+
+		for _, day := range conditions.TimeRange.Days {
+			if !validDays[day] {
+				return fmt.Errorf("invalid day: %s", day)
+			}
+		}
+	}
+
+	// Validate geography
+	for _, geo := range conditions.Geography {
+		if geo == "" {
+			return errors.New("geography cannot be empty")
+		}
+	}
+
+	// Validate customer type
+	for _, ct := range conditions.CustomerType {
+		if ct == "" {
+			return errors.New("customer type cannot be empty")
+		}
+	}
+
+	// Validate volume range
+	if conditions.Volume != nil {
+		if conditions.Volume.Min < 0 || conditions.Volume.Max < 0 {
+			return errors.New("volume range values must be non-negative")
+		}
+
+		if conditions.Volume.Min > conditions.Volume.Max {
+			return errors.New("volume min cannot be greater than max")
+		}
+	}
+
+	// Validate usage pattern
+	if conditions.UsagePattern != nil {
+		if len(conditions.UsagePattern.PeakHours) == 0 && len(conditions.UsagePattern.OffPeakHours) == 0 {
+			return errors.New("usage pattern must have either peak or off-peak hours")
+		}
+
+		// Validate hour formats
+		allHours := append(conditions.UsagePattern.PeakHours, conditions.UsagePattern.OffPeakHours...)
+		for _, hour := range allHours {
+			if !v.isValidHourRange(hour) {
+				return fmt.Errorf("invalid hour range: %s", hour)
+			}
+		}
+	}
+
+	return nil
+}
+
+// ValidateActions validates rule actions
+func (v *PricingValidator) ValidateActions(ctx context.Context, actions pricing.RuleActions) error {
+	if actions.AdjustmentType == "" {
+		return errors.New("adjustment type is required")
+	}
+
+	// Validate adjustment type
+	validTypes := map[pricing.AdjustmentType]bool{
+		pricing.AdjustmentTypePercentage: true,
+		pricing.AdjustmentTypeFixed:      true,
+		pricing.AdjustmentTypeMultiply:   true,
+		pricing.AdjustmentTypeOverride:   true,
+	}
+
+	if !validTypes[actions.AdjustmentType] {
+		return fmt.Errorf("invalid adjustment type: %s", actions.AdjustmentType)
+	}
+
+	// Validate value based on adjustment type
+	switch actions.AdjustmentType {
+	case pricing.AdjustmentTypePercentage:
+		if actions.Value < 0 || actions.Value > 100 {
+			return errors.New("percentage value must be between 0 and 100")
+		}
+
+	case pricing.AdjustmentTypeFixed:
+		if actions.Value < 0 {
+			return errors.New("fixed discount value must be non-negative")
+		}
+
+	case pricing.AdjustmentTypeMultiply:
+		if actions.Value <= 0 {
+			return errors.New("multiplier value must be positive")
+		}
+
+	case pricing.AdjustmentTypeOverride:
+		if actions.NewPrice == nil || *actions.NewPrice < 0 {
+			return errors.New("override action requires a valid new price")
+		}
+	}
+
+	// Validate limit
+	if actions.Limit != nil && *actions.Limit < 0 {
+		return errors.New("limit cannot be negative")
+	}
+
+	return nil
+}
+
+// Helper methods
+
+func (v *PricingValidator) isValidHourRange(hourRange string) bool {
+	// Should be in format "start-end" like "9-17", "18-23", etc.
+	if len(hourRange) < 3 {
+		return false
+	}
+
+	for i, char := range hourRange {
+		if i == 0 && !v.isDigit(char) {
+			return false
+		}
+		if i == len(hourRange)-1 && !v.isDigit(char) {
+			return false
+		}
+		if i == 1 && char != '-' {
+			return false
+		}
+		if i > 1 && i < len(hourRange)-1 && !v.isDigit(char) {
+			return false
+		}
+	}
+
+	return true
+}
+
+func (v *PricingValidator) isDigit(char rune) bool {
+	return char >= '0' && char <= '9'
+}

apps/carrier-connector/internal/services/tenant_analytics_methods.go

@@ -0,0 +1,209 @@
+package services
+
+import (
+	"context"
+	"time"
+
+	"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/tenant"
+)
+
+// parseTimeRange parses time range string and returns start and end dates
+func (s *TenantServiceImpl) parseTimeRange(timeRange string) (time.Time, time.Time) {
+	now := time.Now()
+
+	switch timeRange {
+	case "1h":
+		return now.Add(-1 * time.Hour), now
+	case "24h":
+		return now.Add(-24 * time.Hour), now
+	case "7d":
+		return now.Add(-7 * 24 * time.Hour), now
+	case "30d":
+		return now.Add(-30 * 24 * time.Hour), now
+	case "90d":
+		return now.Add(-90 * 24 * time.Hour), now
+	default:
+		return now.Add(-24 * time.Hour), now
+	}
+}
+
+// buildQuotaStatus builds quota status from usage stats
+func (s *TenantServiceImpl) buildQuotaStatus(usageStats *tenant.TenantUsageStats) []*tenant.TenantUsage {
+	quotaStatus := make([]*tenant.TenantUsage, 0)
+
+	// Create quota status for common resource types
+	resourceTypes := []string{"users", "profiles", "carriers", "api_calls", "storage"}
+
+	for _, resourceType := range resourceTypes {
+		// Get actual usage data from repository
+		usage, err := s.repository.GetUsage(context.Background(), usageStats.TenantID, resourceType)
+		if err != nil {
+			// Create empty usage if not found
+			usage = &tenant.TenantUsage{
+				ID:             generateUsageID(resourceType, usageStats.TenantID),
+				TenantID:       usageStats.TenantID,
+				ResourceType:   resourceType,
+				ResourceCount:  0,
+				QuotaLimit:     0,
+				QuotaUsed:      0,
+				QuotaRemaining: 0,
+				PeriodStart:    time.Now().AddDate(0, -1, 0),
+				PeriodEnd:      time.Now(),
+			}
+		}
+
+		quotaStatus = append(quotaStatus, usage)
+	}
+
+	return quotaStatus
+}
+
+// calculateAPIPerformance calculates real API performance metrics
+func (s *TenantServiceImpl) calculateAPIPerformance(requests []*tenant.APIRequestEvent) *tenant.APIPerformance {
+	if len(requests) == 0 {
+		return &tenant.APIPerformance{
+			TotalRequests:       0,
+			AverageResponseTime: 0,
+			P95ResponseTime:     0,
+			ErrorRate:           0,
+			RequestsPerSecond:   0,
+		}
+	}
+
+	// Calculate performance metrics
+	totalRequests := len(requests)
+	var totalResponseTime int64
+	errorCount := 0
+
+	responseTimes := make([]int, 0, totalRequests)
+
+	for _, req := range requests {
+		totalResponseTime += int64(req.ResponseTime)
+		responseTimes = append(responseTimes, req.ResponseTime)
+
+		if req.StatusCode >= 400 {
+			errorCount++
+		}
+	}
+
+	// Calculate average response time
+	averageResponseTime := float64(totalResponseTime) / float64(totalRequests)
+
+	// Calculate P95 response time
+	sortedResponseTimes := make([]int, len(responseTimes))
+	copy(sortedResponseTimes, responseTimes)
+
+	// Sort response times
+	for i := 0; i < len(sortedResponseTimes); i++ {
+		for j := 0; j < len(sortedResponseTimes)-1-i; j++ {
+			if sortedResponseTimes[j] > sortedResponseTimes[j+1] {
+				sortedResponseTimes[j], sortedResponseTimes[j+1] = sortedResponseTimes[j+1], sortedResponseTimes[j]
+			}
+		}
+	}
+
+	p95Index := int(float64(totalRequests) * 0.95)
+	if p95Index >= totalRequests {
+		p95Index = totalRequests - 1
+	}
+	p95ResponseTime := float64(sortedResponseTimes[p95Index])
+
+	// Calculate error rate
+	errorRate := float64(errorCount) / float64(totalRequests) * 100
+
+	// Calculate requests per second
+	requestsPerSecond := float64(totalRequests) / 3600.0
+
+	return &tenant.APIPerformance{
+		TotalRequests:       totalRequests,
+		AverageResponseTime: averageResponseTime,
+		P95ResponseTime:     p95ResponseTime,
+		ErrorRate:           errorRate,
+		RequestsPerSecond:   requestsPerSecond,
+	}
+}
+
+// parseAPIRequestEvent parses API request event data
+func (s *TenantServiceImpl) parseAPIRequestEvent(event *tenant.TenantEvent) *tenant.APIRequestEvent {
+	endpoint := ""
+	method := "GET"
+	responseTime := 0
+
+	// Extract data from event
+	if eventData, ok := event.EventData["endpoint"]; ok {
+		if ep, ok := eventData.(string); ok {
+			endpoint = ep
+		}
+	}
+
+	if eventData, ok := event.EventData["method"]; ok {
+		if m, ok := eventData.(string); ok {
+			method = m
+		}
+	}
+
+	if eventData, ok := event.EventData["response_time"]; ok {
+		if rt, ok := eventData.(float64); ok {
+			responseTime = int(rt)
+		}
+	}
+
+	statusCode := 200
+	if eventData, ok := event.EventData["status_code"]; ok {
+		if sc, ok := eventData.(float64); ok {
+			statusCode = int(sc)
+		}
+	}
+
+	return &tenant.APIRequestEvent{
+		Timestamp:    event.Timestamp,
+		Endpoint:     endpoint,
+		Method:       method,
+		StatusCode:   statusCode,
+		ResponseTime: responseTime,
+		UserID:       event.UserID,
+	}
+}
+
+// parseErrorEvent parses error event data
+func (s *TenantServiceImpl) parseErrorEvent(event *tenant.TenantEvent) *tenant.ErrorEvent {
+	errorMsg := ""
+
+	if eventData, ok := event.EventData["error"]; ok {
+		if err, ok := eventData.(string); ok {
+			errorMsg = err
+		}
+	}
+
+	return &tenant.ErrorEvent{
+		Timestamp: event.Timestamp,
+		Error:     errorMsg,
+		Context:   event.EventData,
+		UserID:    event.UserID,
+	}
+}
+
+// parseSlowQueryEvent parses slow query event data
+func (s *TenantServiceImpl) parseSlowQueryEvent(event *tenant.TenantEvent) *tenant.SlowQuery {
+	query := ""
+	duration := time.Duration(0)
+
+	if eventData, ok := event.EventData["query"]; ok {
+		if q, ok := eventData.(string); ok {
+			query = q
+		}
+	}
+
+	if eventData, ok := event.EventData["duration"]; ok {
+		if d, ok := eventData.(float64); ok {
+			duration = time.Duration(d) * time.Millisecond
+		}
+	}
+
+	return &tenant.SlowQuery{
+		Timestamp: event.Timestamp,
+		Query:     query,
+		Duration:  duration,
+		Context:   event.EventData,
+	}
+}