Merge pull request #17 from kjannette/poller-tweaks

updated evaluator service -

backend-go/cmd/poller/main.go

@@ -138,6 +138,8 @@ func runCycle(
 		return
 	}
 
+	evaluator.WaitForNotifications()
+
 	duration := time.Since(startTime)
 	log.Printf("[%s] Cycle complete: %d observations, %d alerts fired in %s",
 		time.Now().UTC().Format(time.RFC3339),

backend-go/internal/protocols/ethereum/jsonrpc.go

@@ -18,7 +18,7 @@ import (
 const (
 	rpcTimeoutMS    = 30000
 	rpcMaxRetries   = 3
-	rpcRetryBaseMS  = 1000
+	rpcRetryBaseMS  = 2000
 )
 
 type JsonRpcEthereum struct {

backend-go/internal/services/evaluator.go

@@ -4,8 +4,11 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"sync"
 	"time"
 
+	"golang.org/x/sync/semaphore"
+
 	"github.com/kjannette/koin-ping/backend-go/internal/domain"
 	"github.com/kjannette/koin-ping/backend-go/internal/models"
 	"github.com/kjannette/koin-ping/backend-go/internal/notifications"
@@ -17,6 +20,7 @@ const (
 	notificationTimeout    = 30 * time.Second
 	notificationMaxRetries = 3
 	notificationRetryBase  = time.Second
+	maxConcurrentNotifications = 5
 )
 
 type EvaluatorService struct {
@@ -27,6 +31,8 @@ type EvaluatorService struct {
 	notifConfigs *models.NotificationConfigModel
 	resendAPIKey string
 	emailFrom    string
+	notifSem     *semaphore.Weighted
+	notifWg      sync.WaitGroup
 }
 
 func NewEvaluatorService(
@@ -46,6 +52,7 @@ func NewEvaluatorService(
 		notifConfigs: notifConfigs,
 		resendAPIKey: resendAPIKey,
 		emailFrom:    emailFrom,
+		notifSem:     semaphore.NewWeighted(maxConcurrentNotifications),
 	}
 }
 
@@ -189,7 +196,14 @@ func (s *EvaluatorService) fireAlert(ctx context.Context, rule domain.AlertRule,
 	if addr != nil {
 		userID := addr.UserID
 		address := addr.Address
+		if err := s.notifSem.Acquire(ctx, 1); err != nil {
+			log.Printf("Failed to acquire notification semaphore for rule %d: %v", rule.ID, err)
+			return nil
+		}
+		s.notifWg.Add(1)
 		go func() {
+			defer s.notifSem.Release(1)
+			defer s.notifWg.Done()
 			notifCtx, cancel := context.WithTimeout(context.Background(), notificationTimeout)
 			defer cancel()
 			s.sendNotification(notifCtx, userID, message, obs, addressLabel, rule, address)
@@ -199,6 +213,11 @@ func (s *EvaluatorService) fireAlert(ctx context.Context, rule domain.AlertRule,
 	return nil
 }
 
+// WaitForNotifications blocks until all in-flight notification goroutines finish.
+func (s *EvaluatorService) WaitForNotifications() {
+	s.notifWg.Wait()
+}
+
 func (s *EvaluatorService) buildNotifiers(cfg *domain.NotificationConfig) []notifications.Notifier {
 	var notifiers []notifications.Notifier