feat(discovery): add service skeleton with bucket picker + series matcher

backend/internal/domain/discovery/mock_repo_test.go

@@ -0,0 +1,62 @@
+package discovery
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+)
+
+// Compile-time guard: mockRepo must fully satisfy Repository.
+var _ Repository = (*mockRepo)(nil)
+
+type mockRepo struct {
+	pickStaleFn     func(ctx context.Context, forwardMonths, limit int) ([]Bucket, error)
+	updateBucketFn  func(ctx context.Context, id uuid.UUID, errMsg string) error
+	listSeriesFn    func(ctx context.Context, cityNormalized string) ([]SeriesCandidate, error)
+	editionExistsFn func(ctx context.Context, seriesID uuid.UUID, year int) (bool, error)
+	insertDiscFn    func(ctx context.Context, d DiscoveredMarket) (uuid.UUID, error)
+	isRejectedFn    func(ctx context.Context, nameNormalized, stadt string, year int) (bool, error)
+	queuePendingFn  func(ctx context.Context, nameNormalized, stadt string, startDatum *time.Time) (bool, error)
+}
+
+func (m *mockRepo) PickStaleBuckets(ctx context.Context, fm, lim int) ([]Bucket, error) {
+	return m.pickStaleFn(ctx, fm, lim)
+}
+func (m *mockRepo) UpdateBucketQueried(ctx context.Context, id uuid.UUID, errMsg string) error {
+	return m.updateBucketFn(ctx, id, errMsg)
+}
+func (m *mockRepo) ListSeriesByCity(ctx context.Context, c string) ([]SeriesCandidate, error) {
+	return m.listSeriesFn(ctx, c)
+}
+func (m *mockRepo) EditionExists(ctx context.Context, sid uuid.UUID, y int) (bool, error) {
+	return m.editionExistsFn(ctx, sid, y)
+}
+func (m *mockRepo) InsertDiscovered(ctx context.Context, d DiscoveredMarket) (uuid.UUID, error) {
+	return m.insertDiscFn(ctx, d)
+}
+func (m *mockRepo) IsRejected(ctx context.Context, n, s string, y int) (bool, error) {
+	return m.isRejectedFn(ctx, n, s, y)
+}
+func (m *mockRepo) QueueHasPending(ctx context.Context, n, s string, sd *time.Time) (bool, error) {
+	return m.queuePendingFn(ctx, n, s, sd)
+}
+func (m *mockRepo) ListQueue(ctx context.Context, status string, l, o int) ([]DiscoveredMarket, error) {
+	return nil, nil
+}
+func (m *mockRepo) GetDiscovered(ctx context.Context, id uuid.UUID) (DiscoveredMarket, error) {
+	return DiscoveredMarket{}, nil
+}
+func (m *mockRepo) MarkAccepted(ctx context.Context, tx pgx.Tx, id, eid, r uuid.UUID) error {
+	return nil
+}
+func (m *mockRepo) MarkRejected(ctx context.Context, tx pgx.Tx, id, r uuid.UUID) error {
+	return nil
+}
+func (m *mockRepo) InsertRejection(ctx context.Context, tx pgx.Tx, r RejectedDiscovery) error {
+	return nil
+}
+func (m *mockRepo) BeginTx(ctx context.Context) (pgx.Tx, error) {
+	return nil, nil
+}

backend/internal/domain/discovery/service.go

@@ -0,0 +1,50 @@
+package discovery
+
+import (
+	"context"
+
+	"github.com/google/uuid"
+
+	"marktvogt.de/backend/internal/domain/market"
+)
+
+// Service orchestrates bucket scheduling, agent invocation, and queue management.
+type Service struct {
+	repo          Repository
+	agent         *AgentClient
+	marketService *market.Service
+	batchSize     int
+	forwardMonths int
+}
+
+// NewService constructs a Service. batchSize and forwardMonths configure bucket picking.
+func NewService(repo Repository, agent *AgentClient, marketService *market.Service, batchSize, forwardMonths int) *Service {
+	return &Service{
+		repo:          repo,
+		agent:         agent,
+		marketService: marketService,
+		batchSize:     batchSize,
+		forwardMonths: forwardMonths,
+	}
+}
+
+// PickBuckets returns stale buckets eligible for the next discovery run.
+func (s *Service) PickBuckets(ctx context.Context) ([]Bucket, error) {
+	return s.repo.PickStaleBuckets(ctx, s.forwardMonths, s.batchSize)
+}
+
+// findSeriesMatch returns the ID of the first candidate whose normalized name matches
+// incomingName after normalization. Candidates are expected to be pre-filtered by city.
+func findSeriesMatch(incomingName string, candidates []SeriesCandidate) *uuid.UUID {
+	target := NormalizeName(incomingName)
+	if target == "" {
+		return nil
+	}
+	for _, c := range candidates {
+		if NormalizeName(c.Name) == target {
+			id := c.ID
+			return &id
+		}
+	}
+	return nil
+}

backend/internal/domain/discovery/service_test.go

@@ -0,0 +1,67 @@
+package discovery
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/uuid"
+)
+
+func TestFindSeriesMatch(t *testing.T) {
+	target := SeriesCandidate{
+		ID:   uuid.MustParse("11111111-1111-1111-1111-111111111111"),
+		Name: "Mittelaltermarkt Trostberg",
+		City: "Trostberg",
+	}
+	other := SeriesCandidate{
+		ID:   uuid.MustParse("22222222-2222-2222-2222-222222222222"),
+		Name: "Ritterfest Straßburg",
+		City: "Straßburg",
+	}
+	tests := []struct {
+		name     string
+		incoming string
+		cands    []SeriesCandidate
+		wantID   *uuid.UUID
+	}{
+		{"exact match", "Mittelaltermarkt Trostberg", []SeriesCandidate{target, other}, &target.ID},
+		{"suffix variant matches", "Trostberg Mittelaltermarkt", []SeriesCandidate{target}, &target.ID},
+		{"umlaut match", "Ritterfest Strassburg", []SeriesCandidate{other}, &other.ID},
+		{"no match", "Ganz anderer Markt", []SeriesCandidate{target, other}, nil},
+		{"empty candidates", "Mittelaltermarkt Trostberg", nil, nil},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got := findSeriesMatch(tc.incoming, tc.cands)
+			switch {
+			case tc.wantID == nil && got == nil:
+				// ok
+			case tc.wantID != nil && got != nil && *got == *tc.wantID:
+				// ok
+			default:
+				t.Errorf("got %v, want %v", got, tc.wantID)
+			}
+		})
+	}
+}
+
+func TestPickBucketsPassesConfigToRepo(t *testing.T) {
+	var gotFM, gotLim int
+	m := &mockRepo{
+		pickStaleFn: func(_ context.Context, fm, lim int) ([]Bucket, error) {
+			gotFM, gotLim = fm, lim
+			return []Bucket{{ID: uuid.New(), Land: "Deutschland", Region: "Bayern", YearMonth: "2026-09"}}, nil
+		},
+	}
+	svc := NewService(m, nil, nil, 4, 12)
+	got, err := svc.PickBuckets(context.Background())
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if len(got) != 1 {
+		t.Errorf("got %d buckets, want 1", len(got))
+	}
+	if gotFM != 12 || gotLim != 4 {
+		t.Errorf("forwarded fm=%d lim=%d, want 12,4", gotFM, gotLim)
+	}
+}