diff --git a/apps/api-server/internal/services/subscriber.go b/apps/api-server/internal/services/subscriber.go index 0a09b7b..5495a23 100644 --- a/apps/api-server/internal/services/subscriber.go +++ b/apps/api-server/internal/services/subscriber.go @@ -498,13 +498,96 @@ func (s *SubscriberService) CreateAlert(ctx context.Context, req *models.CreateA // SubscribeToSubscriberUpdates subscribes to subscriber updates func (s *SubscriberService) SubscribeToSubscriberUpdates(ctx context.Context, subscriberId string) (<-chan *models.Subscriber, error) { - // In a real implementation, this would set up WebSocket or pub/sub subscription - // For now, return a channel that closes when context is done - ch := make(chan *models.Subscriber) + ch := make(chan *models.Subscriber, 5) // Buffered channel to prevent blocking + go func() { - <-ctx.Done() - close(ch) + defer close(ch) + + // Poll for subscriber updates every 3 seconds + ticker := time.NewTicker(3 * time.Second) + defer ticker.Stop() + + var lastUpdatedAt time.Time + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + // Query for subscriber updates since last check + var subscriber models.Subscriber + query := s.db.DB.WithContext(ctx).Where("imsi = ?", subscriberId) + if !lastUpdatedAt.IsZero() { + query = query.Where("updated_at > ?", lastUpdatedAt) + } + + if err := query.First(&subscriber).Error; err != nil { + // Subscriber not found or error, continue polling + continue + } + + // Send updated subscriber to channel + select { + case ch <- &subscriber: + lastUpdatedAt = subscriber.UpdatedAt + case <-ctx.Done(): + return + } + } + } }() + + return ch, nil +} + +// SubscribeToAlertUpdates subscribes to alert updates +func (s *SubscriberService) SubscribeToAlertUpdates(ctx context.Context, subscriberId *string) (<-chan *models.Alert, error) { + ch := make(chan *models.Alert, 10) // Buffered channel to prevent blocking + + go func() { + defer close(ch) + + // Poll for new alerts every 2 seconds + ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() + + var lastAlertID uint = 0 + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + // Query for new alerts since last check + var alerts []models.Alert + query := s.db.DB.WithContext(ctx).Where("id > ?", lastAlertID) + + // Filter by subscriber if specified + if subscriberId != nil { + // Convert subscriberId string to uint for database query + if subID, err := strconv.ParseUint(*subscriberId, 10, 32); err == nil { + query = query.Where("subscriber_id = ?", subID) + } + } + + if err := query.Order("id ASC").Limit(10).Find(&alerts).Error; err != nil { + // Log error but continue polling + continue + } + + // Send new alerts to channel + for _, alert := range alerts { + select { + case ch <- &alert: + lastAlertID = alert.ID + case <-ctx.Done(): + return + } + } + } + } + }() + return ch, nil }