From cbfe5a4b9f0af9140d4bd28dced847e9ff7ff321 Mon Sep 17 00:00:00 2001
From: nutcas3 <mauricee423@gmail.com>
Date: Fri, 17 Apr 2026 15:47:47 +0300
Subject: [PATCH] feat: Implement real-time subscription polling for subscriber
 and alert updates

- Update SubscribeToSubscriberUpdates to poll database every 3 seconds for subscriber changes since last update
- Add buffered channel (size 5) to prevent blocking on subscriber updates
- Track lastUpdatedAt timestamp to query only new subscriber changes
- Add SubscribeToAlertUpdates to poll database every 2 seconds for new alerts
- Add buffered channel (size 10) for alert updates with ID-based filtering
- Support optional
---
 .../internal/services/subscriber.go           | 93 ++++++++++++++++++-
 1 file changed, 88 insertions(+), 5 deletions(-)

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
 }