Made the fetcher save what it fetches, with fetcher.WithStore

With WithStore, the Find a Tender and mock workers save each run's
opportunities with SaveOpportunities and log what changed; a failed save
fails the run, so the checkpoint stays put and the next run retries.
Without a store nothing is persisted. The mock's opportunities are saved
with Source "FTSMock" rather than the converter's "FindATender".

fetcher service and job-search serve open the store at database.file
(cli.OpenConfiguredStore), or warn and run without one. A missing file
isn't created, and a database needing migration says to run job-search
migrate.

Fixed upsertOpportunity, which never wrote the source column, so every
saved opportunity had an empty Source.
This commit is contained in:
bwilliams committed 2026-10-16 11:26:16 +01:00
1 parent eba6af680f
commit b3d20c1db0
10 files changed
+407 -38

No files matched your search

+19 -2
View File
@@ -2,6 +2,7 @@ package fetch
import (
"context"
"errors"
"fmt"
"log/slog"
"os"
@@ -25,7 +26,7 @@ func commandService(opts *cli.Options) *cobra.Command {
PersistentPostRun: func(cmd *cobra.Command, args []string) {
slog.Debug("command persistent post-run", "command", "service")
},
RunE: func(cmd *cobra.Command, args []string) error {
RunE: func(cmd *cobra.Command, args []string) (err error) {
slog.Debug("command run", "command", "service")
foreground, _ := cmd.Flags().GetBool("foreground")
@@ -53,7 +54,23 @@ func commandService(opts *cli.Options) *cobra.Command {
stop()
}()
return fetcher.New(&conf.Sources).Run(ctx, since)
var fetcherOptions []fetcher.Option
opportunityStore, openErr := cli.OpenConfiguredStore(ctx, conf)
switch {
case errors.Is(openErr, cli.ErrNoDatabase):
slog.Warn("not storing fetched opportunities", "reason", openErr)
case openErr != nil:
return openErr
default:
defer func() {
err = errors.Join(err, opportunityStore.Close())
}()
fetcherOptions = append(
fetcherOptions, fetcher.WithStore(opportunityStore),
)
}
return fetcher.New(&conf.Sources, fetcherOptions...).Run(ctx, since)
},
}
cmd.Flags().Time(
+21 -4
View File
@@ -29,9 +29,10 @@ func NewCommand(opts *cli.Options) *cobra.Command {
Short: "Serve the REST API (/api) and WebSocket (/ws), and fetch from sources",
Long: `Serve the REST API under /api and the WebSocket endpoint under /ws.
It also runs a worker for every enabled source, as fetcher service does. A
source that stops with an error stops the server too, and the server
stopping stops the sources.`,
It also runs a worker for every enabled source, as fetcher service does,
saving what they fetch to database.file if it is set. A source that stops
with an error stops the server too, and the server stopping stops the
sources.`,
PersistentPreRun: func(cmd *cobra.Command, args []string) {
slog.Debug("command persistent pre-run", "command", "serve")
},
@@ -118,7 +119,23 @@ stopping stops the sources.`,
runCtx, cancelRun := context.WithCancel(ctx)
defer cancelRun()
sourceFetcher := fetcher.New(&conf.Sources)
var fetcherOptions []fetcher.Option
opportunityStore, openErr := cli.OpenConfiguredStore(ctx, conf)
switch {
case errors.Is(openErr, cli.ErrNoDatabase):
slog.Warn("not storing fetched opportunities", "reason", openErr)
case openErr != nil:
return openErr
default:
defer func() {
err = errors.Join(err, opportunityStore.Close())
}()
fetcherOptions = append(
fetcherOptions, fetcher.WithStore(opportunityStore),
)
}
sourceFetcher := fetcher.New(&conf.Sources, fetcherOptions...)
fetcherDone := make(chan error, 1)
go func() {
fetchErr := sourceFetcher.Run(runCtx, time.Time{})
+52
View File
@@ -0,0 +1,52 @@
package cli
import (
"context"
"errors"
"fmt"
"os"
"git.avsdev.uk/Go/job-search/internal/config"
"git.avsdev.uk/Go/job-search/internal/store"
)
// ErrNoDatabase is returned by OpenConfiguredStore when database.file isn't
// set.
var ErrNoDatabase = errors.New("database.file is not set")
// OpenConfiguredStore opens the store at database.file, or returns
// ErrNoDatabase if none is configured. A missing file or a database that
// needs migrating fails with a message saying how to create or migrate it; a
// missing file isn't created, as store.Open would. The caller must close the
// store.
func OpenConfiguredStore(
ctx context.Context, conf *config.JobSearchConfig,
) (*store.Store, error) {
file := conf.Database.File
if file == "" {
return nil, ErrNoDatabase
}
_, err := os.Stat(file)
if errors.Is(err, os.ErrNotExist) {
return nil, fmt.Errorf(
"database %s does not exist: run `job-search migrate up --create`",
file,
)
}
if err != nil {
return nil, fmt.Errorf("database %s: %w", file, err)
}
opportunityStore, err := store.Open(ctx, file)
if errors.Is(err, store.ErrNeedsMigration) {
return nil, fmt.Errorf(
"database %s: %w: run `job-search migrate`", file, err,
)
}
if err != nil {
return nil, fmt.Errorf("database %s: %w", file, err)
}
return opportunityStore, nil
}
+66
View File
@@ -0,0 +1,66 @@
package cli
import (
"context"
"errors"
"os"
"path/filepath"
"strings"
"testing"
"git.avsdev.uk/Go/job-search/internal/config"
"git.avsdev.uk/Go/job-search/internal/store"
)
// No database.file means no store, reported as ErrNoDatabase.
func TestOpenConfiguredStoreNone(t *testing.T) {
opportunityStore, err := OpenConfiguredStore(
context.Background(), &config.JobSearchConfig{},
)
if opportunityStore != nil || !errors.Is(err, ErrNoDatabase) {
t.Errorf(
"OpenConfiguredStore = %v, %v; want nil, ErrNoDatabase",
opportunityStore, err,
)
}
}
// A missing database isn't created; one that needs migrating says how to
// migrate it; a migrated one opens.
func TestOpenConfiguredStore(t *testing.T) {
conf := &config.JobSearchConfig{}
conf.Database.File = filepath.Join(t.TempDir(), "jobsearch.db")
_, err := OpenConfiguredStore(context.Background(), conf)
if err == nil || !strings.Contains(err.Error(), "migrate up --create") {
t.Fatalf("missing: err = %v; want a hint to create it", err)
}
_, statErr := os.Stat(conf.Database.File)
if !errors.Is(statErr, os.ErrNotExist) {
t.Fatal("the missing database file was created")
}
// An empty file is a database at version 0.
if err := os.WriteFile(conf.Database.File, nil, 0o600); err != nil {
t.Fatal(err)
}
_, err = OpenConfiguredStore(context.Background(), conf)
if !errors.Is(err, store.ErrNeedsMigration) ||
!strings.Contains(err.Error(), "run `job-search migrate`") {
t.Fatalf("unmigrated: err = %v; want ErrNeedsMigration and a hint", err)
}
migrated, err := store.Open(
context.Background(), conf.Database.File, store.WithAutoMigrate(),
)
if err != nil {
t.Fatal(err)
}
migrated.Close()
opportunityStore, err := OpenConfiguredStore(context.Background(), conf)
if err != nil {
t.Fatalf("migrated: err = %v", err)
}
opportunityStore.Close()
}
+47 -2
View File
@@ -9,6 +9,8 @@ import (
"time"
"git.avsdev.uk/Go/job-search/internal/config"
"git.avsdev.uk/Go/job-search/internal/store"
"git.avsdev.uk/Go/job-search/internal/types"
"git.avsdev.uk/Go/job-search/internal/util"
)
@@ -27,6 +29,21 @@ func WithLogger(logger *slog.Logger) Option {
}
}
// WithStore makes the workers save the opportunities they fetch to
// opportunityStore. Without it, nothing is persisted.
func WithStore(opportunityStore *store.Store) Option {
return func(f *Fetcher) {
f.store = opportunityStore
}
}
// saveFunc saves the opportunities a worker fetched; see Fetcher.save.
type saveFunc func(
ctx context.Context,
logger *slog.Logger,
opportunities []*types.Opportunity,
) error
// 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.
@@ -48,6 +65,7 @@ func tagged(logger *slog.Logger) *slog.Logger {
type Fetcher struct {
conf *config.JSCSources
log *slog.Logger
store *store.Store // nil: nothing is persisted
started time.Time
// backgroundCtx is the context the workers run under. backgroundCancel
@@ -119,12 +137,12 @@ func (f *Fetcher) Run(ctx context.Context, since time.Time) error {
if f.conf.FindATender.Enabled {
start("FindATender", func(ctx context.Context) error {
return runFindATender(ctx, logger, &f.conf.FindATender, since)
return runFindATender(ctx, logger, &f.conf.FindATender, f.save, since)
})
}
if f.conf.FTSMock.Enabled {
start("FTSMock", func(ctx context.Context) error {
return runFTSMock(ctx, logger, &f.conf.FTSMock, since)
return runFTSMock(ctx, logger, &f.conf.FTSMock, f.save, since)
})
}
if f.conf.Sleeper.Enabled {
@@ -139,3 +157,30 @@ func (f *Fetcher) Run(ctx context.Context, since time.Time) error {
return errors.Join(errs...)
}
// save saves opportunities to the store given WithStore, and logs what
// changed. Without a store it does nothing. A failure fails the worker's run,
// so its checkpoint isn't moved on and the next run fetches them again.
func (f *Fetcher) save(
ctx context.Context,
logger *slog.Logger,
opportunities []*types.Opportunity,
) error {
if f.store == nil {
return nil
}
result, err := f.store.SaveOpportunities(ctx, opportunities)
if err != nil {
return fmt.Errorf("saving opportunities: %w", err)
}
tagged(logger).Info(
"saved opportunities",
"new_opportunities", result.NewOpportunities,
"updated_opportunities", result.UpdatedOpportunities,
"new_releases", result.NewReleases,
)
return nil
}
+162 -12
View File
@@ -15,6 +15,8 @@ import (
"time"
"git.avsdev.uk/Go/job-search/internal/config"
"git.avsdev.uk/Go/job-search/internal/store"
"git.avsdev.uk/Go/job-search/internal/types"
"git.avsdev.uk/Go/job-search/internal/util"
)
@@ -59,20 +61,35 @@ func ftsServer(
}
// fetchFunc fetches from one source at apiRoot.
type fetchFunc func(ctx context.Context, apiRoot string, since time.Time) error
type fetchFunc func(
ctx context.Context, apiRoot string, since time.Time,
) ([]*types.Opportunity, 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}
"FindATender": func(
ctx context.Context, apiRoot string, since time.Time,
) ([]*types.Opportunity, 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}
"FTSMock": func(
ctx context.Context, apiRoot string, since time.Time,
) ([]*types.Opportunity, error) {
conf := &config.JSCSourceFTSMock{
APIRoot: apiRoot,
}
return fetchFTSMock(ctx, slog.Default(), conf, since)
},
}
// noSave is a saveFunc that saves nothing.
func noSave(context.Context, *slog.Logger, []*types.Opportunity) error {
return nil
}
func TestFetchFunctions(t *testing.T) {
tests := []struct {
name string
@@ -102,7 +119,7 @@ func TestFetchFunctions(t *testing.T) {
for _, tc := range tests {
t.Run(fname+"/"+tc.name, func(t *testing.T) {
srv, hits := ftsServer(t, tc.status, tc.body)
err := fetch(
_, err := fetch(
context.Background(), srv.URL, time.Now().Add(-time.Hour),
)
if !tc.check(err) {
@@ -162,7 +179,7 @@ func TestFetchCancelled(t *testing.T) {
cancel()
// Unroutable root: the request should never be waited on.
start := time.Now()
err := fetchers["FTSMock"](ctx, "http://192.0.2.1:9", 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)
}
@@ -196,7 +213,9 @@ 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.JSCSourceSleeper{BlockShutdown: block}
conf := &config.JSCSourceSleeper{
BlockShutdown: block,
}
conf.Interval = time.Hour
ctx, cancel := context.WithCancel(context.Background())
@@ -225,12 +244,16 @@ func TestRunWorkersRejectZeroInterval(t *testing.T) {
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{})
conf := &config.JSCSourceFindATender{
APIRoot: "http://127.0.0.1:1",
}
return runFindATender(ctx, logger, conf, noSave, time.Time{})
},
"FTSMock": func() error {
conf := &config.JSCSourceFTSMock{APIRoot: "http://127.0.0.1:1"}
return runFTSMock(ctx, logger, conf, time.Time{})
conf := &config.JSCSourceFTSMock{
APIRoot: "http://127.0.0.1:1",
}
return runFTSMock(ctx, logger, conf, noSave, time.Time{})
},
"Sleeper": func() error {
return runSleeper(ctx, logger, &config.JSCSourceSleeper{}, time.Time{})
@@ -402,3 +425,130 @@ func TestRunEnabledSources(t *testing.T) {
t.Fatal("Run did not return after cancellation")
}
}
// Each source's opportunities carry its own Source: the mock's are marked
// FTSMock, though its converter marks them as Find a Tender's.
func TestFetchSetsSource(t *testing.T) {
srv, _ := ftsServer(t, http.StatusOK, onePackage)
for name, want := range map[string]string{
"FindATender": "FindATender",
"FTSMock": "FTSMock",
} {
opportunities, err := fetchers[name](
context.Background(), srv.URL, time.Now().Add(-time.Hour),
)
if err != nil {
t.Fatalf("%s: %v", name, err)
}
if len(opportunities) != 1 || opportunities[0].Source != want {
t.Errorf(
"%s: opportunities = %+v; want one from %s",
name, opportunities, want,
)
}
}
}
// openTestStore returns a migrated store in a temporary file.
func openTestStore(t *testing.T) *store.Store {
t.Helper()
path := t.TempDir() + "/fetcher.db"
opportunityStore, err := store.Open(
context.Background(), path, store.WithAutoMigrate(),
)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
opportunityStore.Close()
})
return opportunityStore
}
// WithStore makes the workers save what they fetch, each with its source.
func TestRunSavesToStore(t *testing.T) {
opportunityStore := openTestStore(t)
fat, _ := ftsServer(t, http.StatusOK, onePackage)
mock, _ := ftsServer(
t, http.StatusOK, strings.ReplaceAll(onePackage, "ocds-1", "ocds-2"),
)
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
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() {
done <- New(sources, WithStore(opportunityStore)).Run(
ctx, time.Now().Add(-time.Hour),
)
}()
deadline := time.Now().Add(5 * time.Second)
for {
stored, err := opportunityStore.ListOpportunities(
context.Background(), store.OpportunityFilter{},
)
if err != nil {
t.Fatal(err)
}
if len(stored) == 2 {
break
}
if time.Now().After(deadline) {
t.Fatalf("stored %d opportunities; want 2", len(stored))
}
time.Sleep(5 * time.Millisecond)
}
cancel()
if err := <-done; err != nil {
t.Fatalf("Run = %v", err)
}
for uuid, want := range map[string]string{
"ocds-1": "FindATender",
"ocds-2": "FTSMock",
} {
opportunity, err := opportunityStore.GetOpportunity(
context.Background(), uuid,
)
if err != nil {
t.Fatalf("%s: %v", uuid, err)
}
if opportunity.Source != want {
t.Errorf("%s: Source = %q; want %q", uuid, opportunity.Source, want)
}
}
}
// Without a store, save does nothing; with one that fails, it returns the
// error, so the worker's run fails and its checkpoint stays put.
func TestSave(t *testing.T) {
srv, _ := ftsServer(t, http.StatusOK, onePackage)
opportunities, err := fetchers["FindATender"](
context.Background(), srv.URL, time.Now().Add(-time.Hour),
)
if err != nil {
t.Fatal(err)
}
err = New(&config.JSCSources{}).save(
context.Background(), slog.Default(), opportunities,
)
if err != nil {
t.Errorf("save without a store = %v; want nil", err)
}
closed := openTestStore(t)
closed.Close()
err = New(&config.JSCSources{}, WithStore(closed)).save(
context.Background(), slog.Default(), opportunities,
)
if err == nil || !strings.Contains(err.Error(), "saving opportunities") {
t.Errorf("save to a closed store = %v; want a saving error", err)
}
}
+11 -6
View File
@@ -13,11 +13,12 @@ import (
)
// runFindATender polls Find a Tender every configured interval until ctx
// is cancelled.
// is cancelled, passing what each run fetches to save.
func runFindATender(
ctx context.Context,
logger *slog.Logger,
conf *config.JSCSourceFindATender,
save saveFunc,
since time.Time,
) error {
if conf.APIRoot == "" {
@@ -29,7 +30,11 @@ func runFindATender(
Name: "FindATender",
Interval: conf.Interval,
Task: func(ctx context.Context, since time.Time) error {
return fetchFindATender(ctx, logger, conf, since)
opportunities, err := fetchFindATender(ctx, logger, conf, since)
if err != nil {
return err
}
return save(ctx, logger, opportunities)
},
}
return worker.Run(ctx, since)
@@ -43,7 +48,7 @@ func fetchFindATender(
logger *slog.Logger,
conf *config.JSCSourceFindATender,
since time.Time,
) error {
) ([]*types.Opportunity, error) {
opportunities, err := util.AwaitTask(
ctx,
func() ([]*types.Opportunity, error) {
@@ -55,13 +60,13 @@ func fetchFindATender(
},
)
if errors.Is(err, findatender.ErrNoReleases) {
return ErrNoNewReleases
return nil, ErrNoNewReleases
}
if err != nil {
return err
return nil, err
}
tagged(logger).Info("fetched opportunities", "count", len(opportunities))
return nil
return opportunities, nil
}
+23 -9
View File
@@ -12,12 +12,16 @@ import (
"git.avsdev.uk/Go/job-search/internal/util"
)
// ftsMockSource is the Source of the opportunities the mock source fetches.
const ftsMockSource = "FTSMock"
// runFTSMock polls the mock Find a Tender server every configured interval
// until ctx is cancelled.
// until ctx is cancelled, passing what each run fetches to save.
func runFTSMock(
ctx context.Context,
logger *slog.Logger,
conf *config.JSCSourceFTSMock,
save saveFunc,
since time.Time,
) error {
if conf.APIRoot == "" {
@@ -26,24 +30,30 @@ func runFTSMock(
worker := util.Worker{
Logger: tagged(logger),
Name: "FTSMock",
Name: ftsMockSource,
Interval: conf.Interval,
Task: func(ctx context.Context, since time.Time) error {
return fetchFTSMock(ctx, logger, conf, since)
opportunities, err := fetchFTSMock(ctx, logger, conf, since)
if err != nil {
return err
}
return save(ctx, logger, opportunities)
},
}
return worker.Run(ctx, since)
}
// fetchFTSMock fetches everything updated since since from the mock Find a
// Tender server. Finding nothing returns ErrNoNewReleases, so the worker still
// moves its checkpoint on.
// Tender server. The mock serves the real API's format, so the converter
// marks the opportunities as Find a Tender's; their Source is set to
// ftsMockSource here instead. Finding nothing returns ErrNoNewReleases, so
// the worker still moves its checkpoint on.
func fetchFTSMock(
ctx context.Context,
logger *slog.Logger,
conf *config.JSCSourceFTSMock,
since time.Time,
) error {
) ([]*types.Opportunity, error) {
opportunities, err := util.AwaitTask(
ctx,
func() ([]*types.Opportunity, error) {
@@ -55,13 +65,17 @@ func fetchFTSMock(
},
)
if errors.Is(err, findatender.ErrNoReleases) {
return ErrNoNewReleases
return nil, ErrNoNewReleases
}
if err != nil {
return err
return nil, err
}
for _, opportunity := range opportunities {
opportunity.Source = ftsMockSource
}
tagged(logger).Info("fetched opportunities", "count", len(opportunities))
return nil
return opportunities, nil
}
+3 -2
View File
@@ -32,13 +32,13 @@ const upsertOpportunity = `
INSERT INTO opportunities (
uuid, uuid_type, title, status, description, buyer_name, buyer_id,
main_category, procurement_method, multi_lot, sme_suitable,
vcse_suitable, start_date, end_date, gross_value, has_documents,
vcse_suitable, start_date, end_date, gross_value, has_documents, source,
latest_release_date
) VALUES (
:uuid, :uuid_type, :title, :status, :description, :buyer_name, :buyer_id,
:main_category, :procurement_method, :multi_lot, :sme_suitable,
:vcse_suitable, :start_date, :end_date, :gross_value, :has_documents,
:latest_release_date
:source, :latest_release_date
)
ON CONFLICT (uuid) DO UPDATE SET
uuid_type = excluded.uuid_type,
@@ -56,6 +56,7 @@ ON CONFLICT (uuid) DO UPDATE SET
end_date = excluded.end_date,
gross_value = excluded.gross_value,
has_documents = excluded.has_documents,
source = excluded.source,
latest_release_date = excluded.latest_release_date`
// selectOpportunityByUUID reads one opportunity into a opportunityRow.
+3 -1
View File
@@ -52,6 +52,7 @@ func opportunity(
EndDate: &end,
GrossValue: 125000.50,
HasDocuments: true,
Source: "FindATender",
Classifications: []types.ClassificationCode{
{
IsCPV: true,
@@ -144,7 +145,8 @@ func TestSaveAndGetRoundTrip(t *testing.T) {
got.ProcurementMethod != in.ProcurementMethod || got.UUIDType != "ocid" ||
got.Status != "active" || got.Description != in.Description ||
!got.MultiLot || !got.SMESuitable || got.VCSESuitable ||
!got.HasDocuments || got.GrossValue != in.GrossValue {
!got.HasDocuments || got.GrossValue != in.GrossValue ||
got.Source != in.Source {
t.Errorf("opportunity fields differ:\n got %+v\nwant %+v", got, in)
}
if got.StartDate == nil || !got.StartDate.Equal(day(20)) ||