Made the fetcher a Fetcher type with per-source config, and ran it in job-search serve

fetcher.Run(ctx, conf, since) becomes New(&conf.Sources, opts...) and
Run(ctx, since). Each worker now takes only its own source's config, and
Run starts each one through a closure that binds it. Cancelling ctx stops
the workers through the Fetcher's background context, and Run always waits
for every worker before returning their joined errors.

The findatender client no longer holds the configuration: New takes only
the API root, ReloadConfig is gone, and request tracing follows
logging.IsDebugEnabled(). The FindAOpportunity worker is now FindATender.

job-search serve runs the fetcher alongside the HTTP server. A source that
fails stops the server, and the server stopping stops the sources; with no
sources enabled the server keeps running. fetcher service and fetcher
fts-once use the new APIs.

internal/fetcher/fetcher.go

@@ -16,30 +16,25 @@ import (
 // to fetch; it wraps util.ErrWorkerNoWork.
 var ErrNoNewReleases = fmt.Errorf("%w: no new releases", util.ErrWorkerNoWork)
 
-// Option configures Run.
-type Option func(*options)
-
-// options holds the settings Options change.
-type options struct {
-	log *slog.Logger
-}
+// Option configures a Fetcher.
+type Option func(*Fetcher)
 
 // WithLogger makes Run, its workers and their clients log to logger instead
 // of slog.Default().
 func WithLogger(logger *slog.Logger) Option {
-	return func(opts *options) {
-		opts.log = logger
+	return func(f *Fetcher) {
+		f.log = logger
 	}
 }
 
 // logger returns the logger Run logs to: the given one, or slog.Default().
 // It isn't tagged here, as it is also given to the clients the workers
 // create, which add their own component; see tagged.
-func (opts options) logger() *slog.Logger {
-	if opts.log == nil {
+func (f *Fetcher) logger() *slog.Logger {
+	if f.log == nil {
 		return slog.Default()
 	}
-	return opts.log
+	return f.log
 }
 
 // tagged returns logger tagged with the fetcher's component, for the lines
@@ -48,35 +43,52 @@ func tagged(logger *slog.Logger) *slog.Logger {
 	return logger.With("component", "fetcher")
 }
 
-// runFunc runs one source's worker until ctx is cancelled. logger is
-// untagged: the worker tags its own lines with tagged, and passes logger as
-// it is to any client it creates.
-type runFunc func(
-	ctx context.Context,
-	logger *slog.Logger,
-	conf *config.JobSearchConfig,
-	since time.Time,
-) error
+// Fetcher runs a worker for each enabled source. Create it with New and
+// start it with Run.
+type Fetcher struct {
+	conf    *config.JSCSources
+	log     *slog.Logger
+	started time.Time
+
+	// backgroundCtx is the context the workers run under. backgroundCancel
+	// stops them: Run calls it when its ctx is cancelled, when a worker
+	// fails, and as it returns.
+	backgroundCtx    context.Context
+	backgroundCancel context.CancelFunc
+}
+
+// New returns a Fetcher for the sources conf configures. It logs to
+// slog.Default() unless given WithLogger.
+func New(conf *config.JSCSources, opts ...Option) *Fetcher {
+	fetcher := &Fetcher{
+		conf:    conf,
+		started: time.Now(),
+	}
+	for _, option := range opts {
+		option(fetcher)
+	}
+
+	fetcher.backgroundCtx, fetcher.backgroundCancel = context.WithCancel(
+		context.Background(),
+	)
+
+	return fetcher
+}
 
 // Run starts a worker for each enabled source and blocks until every worker
 // has stopped. Workers stop when ctx is cancelled, or when any worker stops
 // with an error, which starts a shutdown of all of them. The errors are
-// logged as they happen and returned joined; nil means a clean shutdown. It
-// logs to slog.Default() unless given WithLogger.
-func Run(
-	ctx context.Context,
-	conf *config.JobSearchConfig,
-	since time.Time,
-	opts ...Option,
-) error {
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
+// logged as they happen and returned joined; nil means a clean shutdown. Run
+// cancels the Fetcher's background context as it returns, so it can only be
+// called once.
+func (f *Fetcher) Run(ctx context.Context, since time.Time) error {
+	defer f.backgroundCancel()
 
-	var settings options
-	for _, opt := range opts {
-		opt(&settings)
-	}
-	logger := settings.logger()
+	// Cancelling ctx stops the workers, through the background context.
+	stopAfter := context.AfterFunc(ctx, f.backgroundCancel)
+	defer stopAfter()
+
+	logger := f.logger()
 
 	var (
 		wg   sync.WaitGroup
@@ -84,11 +96,14 @@ func Run(
 		errs []error
 	)
 
-	start := func(name string, run runFunc) {
+	// start runs one source's worker until f.backgroundCtx is cancelled. Each
+	// worker is given logger untagged: it tags its own lines with tagged, and
+	// passes logger as it is to any client it creates.
+	start := func(name string, run func(ctx context.Context) error) {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			if err := run(ctx, logger, conf, since); err != nil {
+			if err := run(f.backgroundCtx); err != nil {
 				tagged(logger).Error(
 					"source stopped, shutting down",
 					"source", name,
@@ -97,21 +112,29 @@ func Run(
 				mu.Lock()
 				errs = append(errs, fmt.Errorf("%s: %w", name, err))
 				mu.Unlock()
-				cancel()
+				f.backgroundCancel()
 			}
 		}()
 	}
 
-	if conf.Sources.FindATender.Enabled {
-		start("FindATender", runFindAOpportunity)
+	if f.conf.FindATender.Enabled {
+		start("FindATender", func(ctx context.Context) error {
+			return runFindATender(ctx, logger, &f.conf.FindATender, since)
+		})
 	}
-	if conf.Sources.FTSMock.Enabled {
-		start("FTSMock", runFTSMock)
+	if f.conf.FTSMock.Enabled {
+		start("FTSMock", func(ctx context.Context) error {
+			return runFTSMock(ctx, logger, &f.conf.FTSMock, since)
+		})
 	}
-	if conf.Sources.Sleeper.Enabled {
-		start("Sleeper", runSleeper)
+	if f.conf.Sleeper.Enabled {
+		start("Sleeper", func(ctx context.Context) error {
+			return runSleeper(ctx, logger, &f.conf.Sleeper, since)
+		})
 	}
 
+	// Every worker has stopped once Wait returns, so errs is no longer
+	// written.
 	wg.Wait()
 
 	return errors.Join(errs...)

internal/fetcher/fetcher_test.go

@@ -58,18 +58,22 @@ func ftsServer(
 	return srv, &hits
 }
 
-func confFor(apiRoot string) *config.JobSearchConfig {
-	conf := &config.JobSearchConfig{}
-	conf.Sources.FindATender.APIRoot = apiRoot
-	conf.Sources.FTSMock.APIRoot = apiRoot
-	return conf
+// fetchFunc fetches from one source at apiRoot.
+type fetchFunc func(ctx context.Context, apiRoot string, since time.Time) error
+
+// fetchers are the sources' fetch functions, each given only its API root.
+var fetchers = map[string]fetchFunc{
+	"FindATender": func(ctx context.Context, apiRoot string, since time.Time) error {
+		conf := &config.JSCSourceFindATender{APIRoot: apiRoot}
+		return fetchFindATender(ctx, slog.Default(), conf, since)
+	},
+	"FTSMock": func(ctx context.Context, apiRoot string, since time.Time) error {
+		conf := &config.JSCSourceFTSMock{APIRoot: apiRoot}
+		return fetchFTSMock(ctx, slog.Default(), conf, since)
+	},
 }
 
 func TestFetchFunctions(t *testing.T) {
-	fetchers := map[string]runFunc{
-		"FindAOpportunity": fetchFindAOpportunity,
-		"FTSMock":          fetchFTSMock,
-	}
 	tests := []struct {
 		name    string
 		status  int
@@ -99,10 +103,7 @@ func TestFetchFunctions(t *testing.T) {
 			t.Run(fname+"/"+tc.name, func(t *testing.T) {
 				srv, hits := ftsServer(t, tc.status, tc.body)
 				err := fetch(
-					context.Background(),
-					slog.Default(),
-					confFor(srv.URL),
-					time.Now().Add(-time.Hour),
+					context.Background(), srv.URL, time.Now().Add(-time.Hour),
 				)
 				if !tc.check(err) {
 					t.Fatalf("err = %v, want %s", err, tc.wantErr)
@@ -115,23 +116,44 @@ func TestFetchFunctions(t *testing.T) {
 	}
 }
 
-func TestFetchUsesConfiguredAPIRoot(t *testing.T) {
-	fat, fatHits := ftsServer(t, http.StatusOK, emptyPackage)
-	mock, mockHits := ftsServer(t, http.StatusOK, emptyPackage)
-	conf := &config.JobSearchConfig{}
-	conf.Sources.FindATender.APIRoot = fat.URL
-	conf.Sources.FTSMock.APIRoot = mock.URL
+// Run gives each worker its own source's configuration: with only one source
+// enabled, only its API root is called.
+func TestRunUsesEachSourcesAPIRoot(t *testing.T) {
+	for _, enabled := range []string{"FindATender", "FTSMock"} {
+		fat, fatHits := ftsServer(t, http.StatusOK, emptyPackage)
+		mock, mockHits := ftsServer(t, http.StatusOK, emptyPackage)
+		sources := &config.JSCSources{}
+		sources.FindATender.APIRoot = fat.URL
+		sources.FindATender.Interval = time.Hour
+		sources.FTSMock.APIRoot = mock.URL
+		sources.FTSMock.Interval = time.Hour
+		wanted, other := fatHits, mockHits
+		if enabled == "FindATender" {
+			sources.FindATender.Enabled = true
+		} else {
+			sources.FTSMock.Enabled = true
+			wanted, other = mockHits, fatHits
+		}
 
-	fetchFindAOpportunity(context.Background(), slog.Default(), conf, time.Time{})
-	if fatHits.Load() != 1 || mockHits.Load() != 0 {
-		t.Fatalf(
-			"FindAOpportunity hits: fat=%d mock=%d", fatHits.Load(), mockHits.Load(),
-		)
-	}
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan error, 1)
+		go func() {
+			done <- New(sources).Run(ctx, time.Time{})
+		}()
 
-	fetchFTSMock(context.Background(), slog.Default(), conf, time.Time{})
-	if fatHits.Load() != 1 || mockHits.Load() != 1 {
-		t.Fatalf("FTSMock hits: fat=%d mock=%d", fatHits.Load(), mockHits.Load())
+		deadline := time.Now().Add(5 * time.Second)
+		for wanted.Load() == 0 && time.Now().Before(deadline) {
+			time.Sleep(5 * time.Millisecond)
+		}
+		cancel()
+		<-done
+
+		if wanted.Load() == 0 || other.Load() != 0 {
+			t.Errorf(
+				"%s enabled: its root hit %d times, the other's %d; want >0 and 0",
+				enabled, wanted.Load(), other.Load(),
+			)
+		}
 	}
 }
 
@@ -139,9 +161,8 @@ func TestFetchCancelled(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
 	// Unroutable root: the request should never be waited on.
-	conf := confFor("http://192.0.2.1:9")
 	start := time.Now()
-	err := fetchFTSMock(ctx, slog.Default(), conf, time.Time{})
+	err := fetchers["FTSMock"](ctx, "http://192.0.2.1:9", time.Time{})
 	if !errors.Is(err, context.Canceled) {
 		t.Fatalf("err = %v, want context.Canceled", err)
 	}
@@ -175,9 +196,8 @@ func TestSleepCompletes(t *testing.T) {
 // one holds shutdown open until the worker's shutdown timeout.
 func TestRunSleeperShutdown(t *testing.T) {
 	for _, block := range []bool{false, true} {
-		conf := &config.JobSearchConfig{}
-		conf.Sources.Sleeper.Interval = time.Hour
-		conf.Sources.Sleeper.BlockShutdown = block
+		conf := &config.JSCSourceSleeper{BlockShutdown: block}
+		conf.Interval = time.Hour
 
 		ctx, cancel := context.WithCancel(context.Background())
 		done := make(chan struct{})
@@ -202,17 +222,24 @@ func TestRunSleeperShutdown(t *testing.T) {
 }
 
 func TestRunWorkersRejectZeroInterval(t *testing.T) {
-	conf := &config.JobSearchConfig{}
-	runs := map[string]runFunc{
-		"FindAOpportunity": runFindAOpportunity,
-		"FTSMock":          runFTSMock,
-		"Sleeper":          runSleeper,
+	ctx, logger := context.Background(), slog.Default()
+	runs := map[string]func() error{
+		"FindATender": func() error {
+			conf := &config.JSCSourceFindATender{APIRoot: "http://127.0.0.1:1"}
+			return runFindATender(ctx, logger, conf, time.Time{})
+		},
+		"FTSMock": func() error {
+			conf := &config.JSCSourceFTSMock{APIRoot: "http://127.0.0.1:1"}
+			return runFTSMock(ctx, logger, conf, time.Time{})
+		},
+		"Sleeper": func() error {
+			return runSleeper(ctx, logger, &config.JSCSourceSleeper{}, time.Time{})
+		},
 	}
 	for name, run := range runs {
-		if err := run(
-			context.Background(), slog.Default(), conf, time.Time{},
-		); err == nil {
-			t.Errorf("%s: expected an error for a zero interval", name)
+		err := run()
+		if err == nil || !strings.Contains(err.Error(), "interval") {
+			t.Errorf("%s: err = %v; want an error for a zero interval", name, err)
 		}
 	}
 }
@@ -227,13 +254,13 @@ func TestRunLogsSourceErrors(t *testing.T) {
 	})
 
 	// A zero interval makes each worker fail straight away.
-	conf := &config.JobSearchConfig{}
-	conf.Sources.FindATender.Enabled = true
-	conf.Sources.FindATender.APIRoot = "http://127.0.0.1:1"
-	conf.Sources.FTSMock.Enabled = true
-	conf.Sources.FTSMock.APIRoot = "http://127.0.0.1:1"
-	conf.Sources.Sleeper.Enabled = true
-	err := Run(context.Background(), conf, time.Time{})
+	sources := &config.JSCSources{}
+	sources.FindATender.Enabled = true
+	sources.FindATender.APIRoot = "http://127.0.0.1:1"
+	sources.FTSMock.Enabled = true
+	sources.FTSMock.APIRoot = "http://127.0.0.1:1"
+	sources.Sleeper.Enabled = true
+	err := New(sources).Run(context.Background(), time.Time{})
 
 	for _, name := range []string{"FindATender", "FTSMock", "Sleeper"} {
 		if !strings.Contains(buf.String(), "source="+name) {
@@ -256,24 +283,26 @@ func TestRunMissingAPIRoot(t *testing.T) {
 	quiet := slog.New(slog.DiscardHandler)
 	tests := []struct {
 		name   string
-		enable func(*config.JobSearchConfig)
+		enable func(*config.JSCSources)
 		key    string
 	}{
-		{"find_a_tender", func(conf *config.JobSearchConfig) {
-			conf.Sources.FindATender.Enabled = true
-			conf.Sources.FindATender.Interval = time.Hour
+		{"find_a_tender", func(sources *config.JSCSources) {
+			sources.FindATender.Enabled = true
+			sources.FindATender.Interval = time.Hour
 		}, "sources.find_a_tender.api_root"},
-		{"fts_mock", func(conf *config.JobSearchConfig) {
-			conf.Sources.FTSMock.Enabled = true
-			conf.Sources.FTSMock.Interval = time.Hour
+		{"fts_mock", func(sources *config.JSCSources) {
+			sources.FTSMock.Enabled = true
+			sources.FTSMock.Interval = time.Hour
 		}, "sources.fts_mock.api_root"},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			conf := &config.JobSearchConfig{}
-			tt.enable(conf)
+			sources := &config.JSCSources{}
+			tt.enable(sources)
 
-			err := Run(context.Background(), conf, time.Time{}, WithLogger(quiet))
+			err := New(sources, WithLogger(quiet)).Run(
+				context.Background(), time.Time{},
+			)
 			if err == nil || !strings.Contains(err.Error(), tt.key) {
 				t.Errorf("Run = %v; want an error naming %s", err, tt.key)
 			}
@@ -284,9 +313,8 @@ func TestRunMissingAPIRoot(t *testing.T) {
 func TestRunNoSourcesEnabled(t *testing.T) {
 	done := make(chan struct{})
 	go func() {
-		if err := Run(
-			context.Background(), &config.JobSearchConfig{}, time.Time{},
-		); err != nil {
+		err := New(&config.JSCSources{}).Run(context.Background(), time.Time{})
+		if err != nil {
 			t.Errorf("Run = %v; want nil", err)
 		}
 		close(done)
@@ -303,16 +331,16 @@ func TestRunNoSourcesEnabled(t *testing.T) {
 // cancelled.
 func TestRunShutsDownOnWorkerError(t *testing.T) {
 	mock, mockHits := ftsServer(t, http.StatusOK, emptyPackage)
-	conf := &config.JobSearchConfig{}
-	conf.Sources.FTSMock.Enabled = true
-	conf.Sources.FTSMock.Interval = time.Hour
-	conf.Sources.FTSMock.APIRoot = mock.URL
+	sources := &config.JSCSources{}
+	sources.FTSMock.Enabled = true
+	sources.FTSMock.Interval = time.Hour
+	sources.FTSMock.APIRoot = mock.URL
 	// The sleeper fails straight away: a zero interval is invalid.
-	conf.Sources.Sleeper.Enabled = true
+	sources.Sleeper.Enabled = true
 
 	done := make(chan error, 1)
 	go func() {
-		done <- Run(context.Background(), conf, time.Now().Add(-time.Hour))
+		done <- New(sources).Run(context.Background(), time.Now().Add(-time.Hour))
 	}()
 
 	select {
@@ -337,20 +365,21 @@ func TestRunShutsDownOnWorkerError(t *testing.T) {
 func TestRunEnabledSources(t *testing.T) {
 	fat, fatHits := ftsServer(t, http.StatusOK, onePackage)
 	mock, mockHits := ftsServer(t, http.StatusOK, emptyPackage)
-	conf := &config.JobSearchConfig{}
-	conf.Sources.FindATender.Enabled = true
-	conf.Sources.FindATender.Interval = time.Hour
-	conf.Sources.FindATender.APIRoot = fat.URL
-	conf.Sources.FTSMock.Enabled = true
-	conf.Sources.FTSMock.Interval = time.Hour
-	conf.Sources.FTSMock.APIRoot = mock.URL
-	conf.Sources.Sleeper.Enabled = true
-	conf.Sources.Sleeper.Interval = time.Hour
+	sources := &config.JSCSources{}
+	sources.FindATender.Enabled = true
+	sources.FindATender.Interval = time.Hour
+	sources.FindATender.APIRoot = fat.URL
+	sources.FTSMock.Enabled = true
+	sources.FTSMock.Interval = time.Hour
+	sources.FTSMock.APIRoot = mock.URL
+	sources.Sleeper.Enabled = true
+	sources.Sleeper.Interval = time.Hour
 
 	ctx, cancel := context.WithCancel(context.Background())
 	done := make(chan struct{})
 	go func() {
-		if err := Run(ctx, conf, time.Now().Add(-time.Hour)); err != nil {
+		err := New(sources).Run(ctx, time.Now().Add(-time.Hour))
+		if err != nil {
 			t.Errorf("Run = %v; want nil after cancellation", err)
 		}
 		close(done)

internal/fetcher/worker_findatender.go

@@ -12,43 +12,43 @@ import (
 	"git.avsdev.uk/Go/job-search/internal/util"
 )
 
-// runFindAOpportunity polls Find a Tender every configured interval until ctx
+// runFindATender polls Find a Tender every configured interval until ctx
 // is cancelled.
-func runFindAOpportunity(
+func runFindATender(
 	ctx context.Context,
 	logger *slog.Logger,
-	conf *config.JobSearchConfig,
+	conf *config.JSCSourceFindATender,
 	since time.Time,
 ) error {
-	if conf.Sources.FindATender.APIRoot == "" {
+	if conf.APIRoot == "" {
 		return errors.New("fetcher: sources.find_a_tender.api_root is not set")
 	}
 
 	worker := util.Worker{
 		Logger:   tagged(logger),
-		Name:     "FindAOpportunity",
-		Interval: conf.Sources.FindATender.Interval,
+		Name:     "FindATender",
+		Interval: conf.Interval,
 		Task: func(ctx context.Context, since time.Time) error {
-			return fetchFindAOpportunity(ctx, logger, conf, since)
+			return fetchFindATender(ctx, logger, conf, since)
 		},
 	}
 	return worker.Run(ctx, since)
 }
 
-// fetchFindAOpportunity fetches everything updated since since from Find a
+// fetchFindATender fetches everything updated since since from Find a
 // Tender. Finding nothing returns ErrNoNewReleases, so the worker still moves
 // its checkpoint on.
-func fetchFindAOpportunity(
+func fetchFindATender(
 	ctx context.Context,
 	logger *slog.Logger,
-	conf *config.JobSearchConfig,
+	conf *config.JSCSourceFindATender,
 	since time.Time,
 ) error {
 	opportunities, err := util.AwaitTask(
 		ctx,
 		func() ([]*types.Opportunity, error) {
 			findATender := findatender.New(
-				conf, conf.Sources.FindATender.APIRoot, findatender.WithLogger(logger),
+				conf.APIRoot, findatender.WithLogger(logger),
 			)
 			defer findATender.Shutdown()
 			return findATender.With(ctx).FetchAll(since, true)

internal/fetcher/worker_ftsmock.go

@@ -17,17 +17,17 @@ import (
 func runFTSMock(
 	ctx context.Context,
 	logger *slog.Logger,
-	conf *config.JobSearchConfig,
+	conf *config.JSCSourceFTSMock,
 	since time.Time,
 ) error {
-	if conf.Sources.FTSMock.APIRoot == "" {
+	if conf.APIRoot == "" {
 		return errors.New("fetcher: sources.fts_mock.api_root is not set")
 	}
 
 	worker := util.Worker{
 		Logger:   tagged(logger),
 		Name:     "FTSMock",
-		Interval: conf.Sources.FTSMock.Interval,
+		Interval: conf.Interval,
 		Task: func(ctx context.Context, since time.Time) error {
 			return fetchFTSMock(ctx, logger, conf, since)
 		},
@@ -41,14 +41,14 @@ func runFTSMock(
 func fetchFTSMock(
 	ctx context.Context,
 	logger *slog.Logger,
-	conf *config.JobSearchConfig,
+	conf *config.JSCSourceFTSMock,
 	since time.Time,
 ) error {
 	opportunities, err := util.AwaitTask(
 		ctx,
 		func() ([]*types.Opportunity, error) {
 			findATender := findatender.New(
-				conf, conf.Sources.FTSMock.APIRoot, findatender.WithLogger(logger),
+				conf.APIRoot, findatender.WithLogger(logger),
 			)
 			defer findATender.Shutdown()
 			return findATender.With(ctx).FetchAll(since, true)

internal/fetcher/worker_sleeper.go

@@ -17,14 +17,14 @@ const sleepDuration = 60 * time.Second
 func runSleeper(
 	ctx context.Context,
 	logger *slog.Logger,
-	conf *config.JobSearchConfig,
+	conf *config.JSCSourceSleeper,
 	since time.Time,
 ) error {
-	blockShutdown := conf.Sources.Sleeper.BlockShutdown
+	blockShutdown := conf.BlockShutdown
 	worker := util.Worker{
 		Logger:   tagged(logger),
 		Name:     "Sleeper",
-		Interval: conf.Sources.Sleeper.Interval,
+		Interval: conf.Interval,
 		Task: func(taskCtx context.Context, since time.Time) error {
 			if blockShutdown {
 				return sleep(taskCtx, true, sleepDuration)