Add background IP intel and restore dashboard stats

2026-03-12 11:10:59 +01:00
parent 14a711038b
commit 1822e2148a
9 changed files with 506 additions and 31 deletions
--- a/internal/service/service.go
+++ b/internal/service/service.go
@@ -29,6 +29,10 @@ type Service struct {
 	blocker      opnsense.AliasClient
 	investigator Investigator
 	logger       *log.Logger
+
+	investigationQueueMu sync.Mutex
+	investigationQueued  map[string]struct{}
+	investigationQueue   chan string
 }

 type Investigator interface {
@@ -39,7 +43,7 @@ func New(cfg *config.Config, db *store.Store, blocker opnsense.AliasClient, inve
 	if logger == nil {
 		logger = log.New(io.Discard, "", 0)
 	}
-	return &Service{
+	service := &Service{
 		cfg:          cfg,
 		store:        db,
 		evaluator:    engine.NewEvaluator(),
@@ -47,10 +51,33 @@ func New(cfg *config.Config, db *store.Store, blocker opnsense.AliasClient, inve
 		investigator: investigator,
 		logger:       logger,
 	}
+	if investigator != nil && cfg.Investigation.BackgroundWorkers > 0 {
+		queueSize := cfg.Investigation.BackgroundBatchSize
+		if queueSize < 64 {
+			queueSize = 64
+		}
+		service.investigationQueue = make(chan string, queueSize)
+		service.investigationQueued = make(map[string]struct{}, queueSize)
+	}
+	return service
 }

 func (s *Service) Run(ctx context.Context) error {
 	var wg sync.WaitGroup
+	if s.investigationQueue != nil {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			s.runInvestigationScheduler(ctx)
+		}()
+		for workerIndex := 0; workerIndex < s.cfg.Investigation.BackgroundWorkers; workerIndex++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				s.runInvestigationWorker(ctx)
+			}()
+		}
+	}
 	for _, source := range s.cfg.Sources {
 		source := source
 		wg.Add(1)
@@ -81,12 +108,25 @@ func (s *Service) ListRecentIPs(ctx context.Context, since time.Time, limit int)
 	if err != nil {
 		return nil, err
 	}
+	investigations, err := s.store.GetInvestigationsForIPs(ctx, recentRowIPs(items))
+	if err != nil {
+		return nil, err
+	}
+	staleSince := time.Now().UTC().Add(-s.cfg.Investigation.RefreshAfter.Duration)
 	for index := range items {
 		state := model.IPState{
 			IP:             items[index].IP,
 			State:          items[index].State,
 			ManualOverride: items[index].ManualOverride,
 		}
+		if investigation, ok := investigations[items[index].IP]; ok {
+			items[index].Bot = investigation.Bot
+			if investigation.UpdatedAt.Before(staleSince) {
+				s.enqueueInvestigation(items[index].IP)
+			}
+		} else {
+			s.enqueueInvestigation(items[index].IP)
+		}
 		backend := s.resolveOPNsenseStatus(ctx, state)
 		items[index].Actions = actionAvailability(state, backend)
 	}
@@ -114,22 +154,12 @@ func (s *Service) InvestigateIP(ctx context.Context, ip string) (model.IPDetails
 	if err != nil {
 		return model.IPDetails{}, err
 	}
-	if s.investigator != nil {
-		investigation, found, err := s.store.GetInvestigation(ctx, normalized)
-		if err != nil {
-			return model.IPDetails{}, err
-		}
-		shouldRefresh := !found || time.Since(investigation.UpdatedAt) >= s.cfg.Investigation.RefreshAfter.Duration
-		if shouldRefresh {
-			fresh, err := s.investigator.Investigate(ctx, normalized, collectUserAgents(details.RecentEvents))
-			if err != nil {
-				return model.IPDetails{}, err
-			}
-			if err := s.store.SaveInvestigation(ctx, fresh); err != nil {
-				return model.IPDetails{}, err
-			}
-			details.Investigation = &fresh
-		}
+	fresh, err := s.refreshInvestigation(ctx, normalized, true)
+	if err != nil {
+		return model.IPDetails{}, err
+	}
+	if fresh != nil {
+		details.Investigation = fresh
 	}
 	return s.decorateDetails(ctx, details)
 }
@@ -281,9 +311,125 @@ func (s *Service) processRecord(ctx context.Context, source config.SourceConfig,
 			return err
 		}
 	}
+	s.enqueueInvestigation(record.ClientIP)
 	return nil
 }

+func (s *Service) runInvestigationScheduler(ctx context.Context) {
+	s.enqueueRecentInvestigations(ctx)
+	ticker := time.NewTicker(s.cfg.Investigation.BackgroundPollInterval.Duration)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			s.enqueueRecentInvestigations(ctx)
+		}
+	}
+}
+
+func (s *Service) runInvestigationWorker(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case ip := <-s.investigationQueue:
+			func() {
+				defer s.markInvestigationDone(ip)
+				workerCtx, cancel := context.WithTimeout(ctx, s.cfg.Investigation.Timeout.Duration)
+				_, err := s.refreshInvestigation(workerCtx, ip, false)
+				cancel()
+				if err != nil && !errors.Is(err, context.Canceled) {
+					s.logger.Printf("investigation %s: %v", ip, err)
+				}
+			}()
+		}
+	}
+}
+
+func (s *Service) enqueueRecentInvestigations(ctx context.Context) {
+	if s.investigationQueue == nil {
+		return
+	}
+	since := time.Now().UTC().Add(-s.cfg.Investigation.BackgroundLookback.Duration)
+	items, err := s.store.ListRecentIPRows(ctx, since, s.cfg.Investigation.BackgroundBatchSize)
+	if err != nil {
+		s.logger.Printf("list recent IPs for investigation: %v", err)
+		return
+	}
+	investigations, err := s.store.GetInvestigationsForIPs(ctx, recentRowIPs(items))
+	if err != nil {
+		s.logger.Printf("list investigations for recent IPs: %v", err)
+		return
+	}
+	staleSince := time.Now().UTC().Add(-s.cfg.Investigation.RefreshAfter.Duration)
+	for _, item := range items {
+		investigation, found := investigations[item.IP]
+		if !found || investigation.UpdatedAt.Before(staleSince) {
+			s.enqueueInvestigation(item.IP)
+		}
+	}
+}
+
+func (s *Service) enqueueInvestigation(ip string) {
+	if s.investigationQueue == nil {
+		return
+	}
+	normalized, err := normalizeIP(ip)
+	if err != nil {
+		return
+	}
+	s.investigationQueueMu.Lock()
+	if _, ok := s.investigationQueued[normalized]; ok {
+		s.investigationQueueMu.Unlock()
+		return
+	}
+	s.investigationQueued[normalized] = struct{}{}
+	s.investigationQueueMu.Unlock()
+	select {
+	case s.investigationQueue <- normalized:
+	default:
+		s.markInvestigationDone(normalized)
+	}
+}
+
+func (s *Service) markInvestigationDone(ip string) {
+	s.investigationQueueMu.Lock()
+	defer s.investigationQueueMu.Unlock()
+	delete(s.investigationQueued, ip)
+}
+
+func (s *Service) refreshInvestigation(ctx context.Context, ip string, force bool) (*model.IPInvestigation, error) {
+	if s.investigator == nil {
+		return nil, nil
+	}
+	normalized, err := normalizeIP(ip)
+	if err != nil {
+		return nil, err
+	}
+	investigation, found, err := s.store.GetInvestigation(ctx, normalized)
+	if err != nil {
+		return nil, err
+	}
+	shouldRefresh := force || !found || time.Since(investigation.UpdatedAt) >= s.cfg.Investigation.RefreshAfter.Duration
+	if !shouldRefresh {
+		return &investigation, nil
+	}
+	userAgents, err := s.store.ListRecentUserAgentsForIP(ctx, normalized, 12)
+	if err != nil {
+		return nil, err
+	}
+	fresh, err := s.investigator.Investigate(ctx, normalized, userAgents)
+	if err != nil {
+		return nil, err
+	}
+	if err := s.store.SaveInvestigation(ctx, fresh); err != nil {
+		return nil, err
+	}
+	return &fresh, nil
+}
+
 func (s *Service) readNewLines(ctx context.Context, source config.SourceConfig) ([]string, error) {
 	info, err := os.Stat(source.Path)
 	if err != nil {
@@ -521,3 +667,11 @@ func collectUserAgents(events []model.Event) []string {
 	}
 	return items
 }
+
+func recentRowIPs(items []model.RecentIPRow) []string {
+	result := make([]string, 0, len(items))
+	for _, item := range items {
+		result = append(result, item.IP)
+	}
+	return result
+}