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
This commit is contained in:
nutcas3
2026-04-17 15:47:47 +03:00
parent 9d2e5b7175
commit cbfe5a4b9f
@@ -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
}