qc-text-builder/internal/polling/service.go

package polling

import (
	"context"
	"log/slog"
	"sync"
	"time"

	"qctextbuilder/internal/buildsvc"
	"qctextbuilder/internal/store"
)

type Service struct {
	buildSvc buildsvc.Service
	store    store.BuildStore
	interval time.Duration
	maxPolls int
	logger   *slog.Logger
}

func New(buildSvc buildsvc.Service, buildStore store.BuildStore, interval time.Duration, maxPolls int, logger *slog.Logger) *Service {
	if interval <= 0 {
		interval = 5 * time.Second
	}
	if maxPolls <= 0 {
		maxPolls = 4
	}
	return &Service{
		buildSvc: buildSvc,
		store:    buildStore,
		interval: interval,
		maxPolls: maxPolls,
		logger:   logger,
	}
}

func (s *Service) Start(ctx context.Context) error {
	ticker := time.NewTicker(s.interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
			s.PollPendingOnce(ctx, 20)
		}
	}
}

func (s *Service) PollPendingOnce(ctx context.Context, limit int) {
	builds, err := s.store.ListBuildsByStatuses(ctx, []string{"queued", "processing"}, limit)
	if err != nil {
		s.logger.Error("list pending builds failed", "error", err)
		return
	}

	semSize := s.maxPolls
	if semSize <= 0 {
		semSize = 1
	}
	sem := make(chan struct{}, semSize)
	var wg sync.WaitGroup

	for _, build := range builds {
		select {
		case <-ctx.Done():
			return
		default:
		}

		wg.Add(1)
		sem <- struct{}{}
		go func(buildID string) {
			defer wg.Done()
			defer func() { <-sem }()
			if err := s.buildSvc.PollOnce(ctx, buildID); err != nil {
				s.logger.Error("poll build failed", "buildId", buildID, "error", err)
			}
		}(build.ID)
	}

	wg.Wait()
}