general cleanup - removed old comments, enforced naming conventions etc

backend/internal/services/evaluator.go

@@ -0,0 +1,363 @@
+package services
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"golang.org/x/sync/semaphore"
+
+	"github.com/kjannette/koin-ping/backend/internal/domain"
+	"github.com/kjannette/koin-ping/backend/internal/models"
+	"github.com/kjannette/koin-ping/backend/internal/notifications"
+	"github.com/kjannette/koin-ping/backend/internal/protocols/ethereum"
+	"github.com/kjannette/koin-ping/backend/internal/wei"
+)
+
+const (
+	notificationTimeout    = 30 * time.Second
+	notificationMaxRetries = 3
+	notificationRetryBase  = time.Second
+	maxConcurrentNotifications = 5
+)
+
+type EvaluatorService struct {
+	eth          ethereum.EthereumObserver
+	alertRules   *models.AlertRuleModel
+	alertEvents  *models.AlertEventModel
+	addresses    *models.AddressModel
+	notifConfigs *models.NotificationConfigModel
+	resendAPIKey string
+	emailFrom    string
+	notifSem     *semaphore.Weighted
+	notifWg      sync.WaitGroup
+}
+
+func NewEvaluatorService(
+	eth ethereum.EthereumObserver,
+	alertRules *models.AlertRuleModel,
+	alertEvents *models.AlertEventModel,
+	addresses *models.AddressModel,
+	notifConfigs *models.NotificationConfigModel,
+	resendAPIKey string,
+	emailFrom string,
+) *EvaluatorService {
+	return &EvaluatorService{
+		eth:          eth,
+		alertRules:   alertRules,
+		alertEvents:  alertEvents,
+		addresses:    addresses,
+		notifConfigs: notifConfigs,
+		resendAPIKey: resendAPIKey,
+		emailFrom:    emailFrom,
+		notifSem:     semaphore.NewWeighted(maxConcurrentNotifications),
+	}
+}
+
+func (s *EvaluatorService) Evaluate(ctx context.Context, observations []domain.ObservedTx) (int, error) {
+	alertsFired := 0
+
+	for _, obs := range observations {
+		fired, err := s.evaluateObservation(ctx, obs)
+		if err != nil {
+			log.Printf("Error evaluating observation for address ID %d: %v", obs.AddressID, err)
+			continue
+		}
+		alertsFired += fired
+	}
+
+	return alertsFired, nil
+}
+
+func (s *EvaluatorService) evaluateObservation(ctx context.Context, obs domain.ObservedTx) (int, error) {
+	rules, err := s.alertRules.ListByAddress(ctx, obs.AddressID)
+	if err != nil {
+		return 0, err
+	}
+
+	alertsFired := 0
+	for _, rule := range rules {
+		if !rule.Enabled {
+			continue
+		}
+
+		matches, err := s.ruleMatches(ctx, rule, obs)
+		if err != nil {
+			log.Printf("Error matching rule %d: %v", rule.ID, err)
+			continue
+		}
+
+		if matches {
+			if err := s.fireAlert(ctx, rule, obs); err != nil {
+				log.Printf("Error firing alert for rule %d: %v", rule.ID, err)
+				continue
+			}
+			alertsFired++
+		}
+	}
+
+	return alertsFired, nil
+}
+
+func (s *EvaluatorService) ruleMatches(ctx context.Context, rule domain.AlertRule, obs domain.ObservedTx) (bool, error) {
+	switch rule.Type {
+	case domain.AlertIncomingTx:
+		return obs.Direction == domain.DirectionIncoming, nil
+	case domain.AlertOutgoingTx:
+		return obs.Direction == domain.DirectionOutgoing, nil
+	case domain.AlertLargeTransfer:
+		return s.matchesLargeTransfer(rule, obs)
+	case domain.AlertBalanceBelow:
+		return s.matchesBalanceBelow(ctx, rule, obs)
+	default:
+		log.Printf("Unknown rule type: %s", rule.Type)
+		return false, nil
+	}
+}
+
+func (s *EvaluatorService) matchesLargeTransfer(rule domain.AlertRule, obs domain.ObservedTx) (bool, error) {
+	if rule.Threshold == nil {
+		return false, nil
+	}
+
+	thresholdWei, err := wei.FromEth(*rule.Threshold)
+	if err != nil {
+		return false, err
+	}
+
+	return wei.GreaterThanOrEqual(obs.Value, thresholdWei)
+}
+
+// matchesBalanceBelow only triggers after outgoing transactions, since incoming
+// transactions increase balance and can't cause it to drop below threshold.
+func (s *EvaluatorService) matchesBalanceBelow(ctx context.Context, rule domain.AlertRule, obs domain.ObservedTx) (bool, error) {
+	if rule.Threshold == nil {
+		return false, nil
+	}
+
+	if obs.Direction != domain.DirectionOutgoing {
+		return false, nil
+	}
+
+	addr, err := s.addresses.FindByID(ctx, obs.AddressID, nil)
+	if err != nil {
+		return false, err
+	}
+	if addr == nil {
+		log.Printf("Address ID %d not found", obs.AddressID)
+		return false, nil
+	}
+
+	balanceWei, err := s.eth.GetBalance(ctx, addr.Address)
+	if err != nil {
+		return false, err
+	}
+
+	thresholdWei, err := wei.FromEth(*rule.Threshold)
+	if err != nil {
+		return false, err
+	}
+
+	return wei.LessThan(balanceWei, thresholdWei)
+}
+
+func (s *EvaluatorService) fireAlert(ctx context.Context, rule domain.AlertRule, obs domain.ObservedTx) error {
+	addr, err := s.addresses.FindByID(ctx, obs.AddressID, nil)
+	if err != nil {
+		return err
+	}
+
+	addressLabel := "Unknown"
+	if addr != nil {
+		if addr.Label != nil {
+			addressLabel = *addr.Label
+		} else {
+			addressLabel = addr.Address
+		}
+	}
+
+	message := s.buildMessage(rule, obs)
+	txHash := &obs.Hash
+
+	event, err := s.alertEvents.Create(ctx, rule.ID, message, &addressLabel, txHash)
+	if err != nil {
+		return err
+	}
+
+	if event == nil {
+		log.Printf("[ALERT DEDUP] Rule %d (%s) - duplicate event skipped for TX: %s", rule.ID, rule.Type, obs.Hash)
+		return nil
+	}
+
+	log.Printf("[ALERT FIRED] Rule %d (%s) - %s - TX: %s", rule.ID, rule.Type, message, obs.Hash)
+
+	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)
+		}()
+	}
+
+	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
+
+	if cfg.DiscordWebhookURL != nil && *cfg.DiscordWebhookURL != "" {
+		notifiers = append(notifiers, &notifications.DiscordNotifier{WebhookURL: *cfg.DiscordWebhookURL})
+	}
+
+	if cfg.TelegramBotToken != nil && *cfg.TelegramBotToken != "" &&
+		cfg.TelegramChatID != nil && *cfg.TelegramChatID != "" {
+		notifiers = append(notifiers, &notifications.TelegramNotifier{
+			BotToken: *cfg.TelegramBotToken,
+			ChatID:   *cfg.TelegramChatID,
+		})
+	}
+
+	if cfg.SlackWebhookURL != nil && *cfg.SlackWebhookURL != "" {
+		notifiers = append(notifiers, &notifications.SlackNotifier{WebhookURL: *cfg.SlackWebhookURL})
+	}
+
+	if cfg.Email != nil && *cfg.Email != "" {
+		notifiers = append(notifiers, &notifications.EmailNotifier{
+			APIKey: s.resendAPIKey,
+			From:   s.emailFrom,
+			To:     *cfg.Email,
+		})
+	}
+
+	return notifiers
+}
+
+func sendWithRetry(ctx context.Context, n notifications.Notifier, message string, meta notifications.AlertMetadata) error {
+	var lastErr error
+	for attempt := range notificationMaxRetries {
+		if attempt > 0 {
+			wait := notificationRetryBase * time.Duration(1<<(attempt-1))
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-time.After(wait):
+			}
+		}
+
+		if err := n.Send(ctx, message, meta); err != nil {
+			log.Printf("Notification attempt %d/%d failed: %v", attempt+1, notificationMaxRetries, err)
+			lastErr = err
+			continue
+		}
+
+		return nil
+	}
+
+	return lastErr
+}
+
+func (s *EvaluatorService) sendNotification(ctx context.Context, userID, message string, obs domain.ObservedTx, addressLabel string, rule domain.AlertRule, address string) {
+	notifConfig, err := s.notifConfigs.GetConfig(ctx, userID)
+	if err != nil {
+		log.Printf("Failed to get notification config: %v", err)
+		return
+	}
+
+	if notifConfig == nil || !notifConfig.NotificationEnabled {
+		return
+	}
+
+	meta := notifications.AlertMetadata{
+		TxHash:       obs.Hash,
+		AddressLabel: addressLabel,
+		AlertType:    string(rule.Type),
+		Address:      address,
+	}
+
+	for _, n := range s.buildNotifiers(notifConfig) {
+		if err := sendWithRetry(ctx, n, message, meta); err != nil {
+			log.Printf("Notification channel failed for user %s after retries: %v", userID, err)
+		} else {
+			log.Printf("Notification sent to user %s via %T", userID, n)
+		}
+	}
+}
+
+func (s *EvaluatorService) buildMessage(rule domain.AlertRule, obs domain.ObservedTx) string {
+	if obs.IsTokenTransfer() {
+		return s.buildTokenMessage(rule, obs)
+	}
+
+	switch rule.Type {
+	case domain.AlertIncomingTx:
+		ethStr, _ := wei.FormatAsEth(obs.Value, 4)
+		return fmt.Sprintf("Incoming transaction: %s received", ethStr)
+	case domain.AlertOutgoingTx:
+		ethStr, _ := wei.FormatAsEth(obs.Value, 4)
+		return fmt.Sprintf("Outgoing transaction: %s sent", ethStr)
+	case domain.AlertLargeTransfer:
+		ethStr, _ := wei.FormatAsEth(obs.Value, 4)
+		threshold := float64(0)
+		if rule.Threshold != nil {
+			threshold = *rule.Threshold
+		}
+		return fmt.Sprintf("Large transfer detected: %s (threshold: %g ETH)", ethStr, threshold)
+	case domain.AlertBalanceBelow:
+		threshold := float64(0)
+		if rule.Threshold != nil {
+			threshold = *rule.Threshold
+		}
+		return fmt.Sprintf("Balance dropped below threshold of %g ETH", threshold)
+	default:
+		return "Alert triggered"
+	}
+}
+
+const defaultTokenDecimals = 18
+
+func (s *EvaluatorService) buildTokenMessage(rule domain.AlertRule, obs domain.ObservedTx) string {
+	symbol := "tokens"
+	if obs.TokenSymbol != nil {
+		symbol = *obs.TokenSymbol
+	}
+
+	amount := "unknown"
+	if obs.TokenValue != nil {
+		decimals := defaultTokenDecimals
+		if obs.TokenDecimals != nil {
+			decimals = *obs.TokenDecimals
+		}
+		amount = wei.FormatTokenAmount(*obs.TokenValue, decimals)
+	}
+
+	switch rule.Type {
+	case domain.AlertIncomingTx:
+		return fmt.Sprintf("Incoming transfer: %s %s received", amount, symbol)
+	case domain.AlertOutgoingTx:
+		return fmt.Sprintf("Outgoing transfer: %s %s sent", amount, symbol)
+	case domain.AlertLargeTransfer:
+		threshold := float64(0)
+		if rule.Threshold != nil {
+			threshold = *rule.Threshold
+		}
+		return fmt.Sprintf("Large token transfer: %s %s (threshold: %g)", amount, symbol, threshold)
+	default:
+		return fmt.Sprintf("Token transfer: %s %s", amount, symbol)
+	}
+}