From 284a5896279e3fce2005014f83ed477ca270ec62 Mon Sep 17 00:00:00 2001 From: Craig Williams Date: Fri, 16 Oct 2026 11:37:09 +0100 Subject: [PATCH] Made each source start from its newest stored release, with --since as an override Without --since, fetcher service and job-search serve start each source from the newest release stored for it (the new store.LatestReleaseDate, by Source), so a restart carries on where it left off instead of fetching the last day again; with nothing stored, the client's default still applies. --since overrides it for every source. The Find a Tender source's name is now a constant, as the Source its opportunities are stored with. The README's guide to adding a source is brought up to date with the current worker, start and config type names. --- README.md | 35 +++++--- internal/cli/fetch/service.go | 3 +- internal/cli/serve/command.go | 3 +- internal/fetcher/fetcher.go | 57 +++++++++++-- internal/fetcher/fetcher_test.go | 107 +++++++++++++++++++++++++ internal/fetcher/worker_findatender.go | 6 +- internal/store/queries.go | 10 +++ internal/store/queries_test.go | 3 + internal/store/store.go | 20 +++++ internal/store/store_test.go | 34 ++++++++ 10 files changed, 256 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 0ccb3d8..26dfe81 100644 --- a/README.md +++ b/README.md @@ -84,6 +84,10 @@ refuses to open a database that isn't at this build's schema version, so run `service` logs to `logging.file` if it's set, or to the terminal with `--foreground`. Both commands take `--since "YYYY-MM-DD[ hh:mm:ss]"`. +Without it, `service` (and `job-search serve`) starts each source from the +newest release stored for it, so a restart carries on where it left off; with +nothing stored, Find a Tender's client fetches the last day. `--since` +overrides that for every source. `fts-once` can be limited with `--planning`, `--opportunities` or `--awards`, and with `--ignore-rate-limit` it fails on a rate limit instead of waiting. @@ -169,7 +173,7 @@ Using a source called `example` as an illustration: 1. **Configuration section:** create `internal/config/conf_source_example.go`, copying an existing source such as `conf_source_ftsmock.go`: - Add a `JSCKSourceExample = "example"` constant. - - Define `JobSearchConfigSourceExample`, embedding `JobSearchConfigSource` + - Define `JSCSourceExample`, embedding `JSCSourcesSource` (for `enabled` and `interval`), plus any fields of its own with `cfg` and optional `default` tags, e.g. ``APIRoot string `cfg:"api_root"` ``. - Implement the `Section` interface (see `internal/config/section.go`): @@ -178,8 +182,8 @@ Using a source called `example` as an illustration: `// ---------- Section methods (see Section) ----------` banner, renaming the receiver type and using the new constant in `Flatten`. 2. **Register the section:** add - ``Example JobSearchConfigSourceExample `cfg:"example"` `` to - `JobSearchConfigSources` in `internal/config/conf_sources.go`. The `cfg` tag + ``Example JSCSourceExample `cfg:"example"` `` to + `JSCSources` in `internal/config/conf_sources.go`. The `cfg` tag must match the constant. The source's keys (`sources.example.*`) are then loaded, validated and available to the `config` commands with no further code. @@ -189,18 +193,27 @@ Using a source called `example` as an illustration: for `fts_mock`: create it with the source's own API root. 4. **Worker:** create `internal/fetcher/worker_example.go`, following `worker_ftsmock.go`: - - `runExample(ctx, conf, since)` builds a `util.Worker` with the source's - name, `conf.Sources.Example.Interval` and a task, then calls `Run`. - - The task fetches everything since its `since` argument. Wrap blocking - calls in `util.AwaitTask` so shutdown isn't held up, and return - `ErrNoNewReleases` when nothing is found, so the checkpoint still moves - forward. Any other error makes the next run retry from the same point. + - Add an `exampleSource = "Example"` constant: the worker's name, and the + `Source` its opportunities are stored with (set it on each one if the + client doesn't, as `fetchFTSMock` does). + - `runExample(ctx, logger, conf, save, since)` builds a `util.Worker` + with `exampleSource`, `conf.Interval` and a task, then calls `Run`. + - The task fetches everything since its `since` argument, then returns + `save(ctx, logger, opportunities)`, which stores them when the fetcher + has a store. Wrap blocking calls in `util.AwaitTask` so shutdown isn't + held up, and return `ErrNoNewReleases` when nothing is found, so the + checkpoint still moves forward. Any other error makes the next run + retry from the same point. 5. **Start it:** in `Run` in `internal/fetcher/fetcher.go`, add ```go - if conf.Sources.Example.Enabled { - start("Example", runExample) + if f.conf.Example.Enabled { + start(exampleSource, func(ctx context.Context, since time.Time) error { + return runExample(ctx, logger, &f.conf.Example, f.save, since) + }) } ``` + `start` passes the worker its starting point: `--since`, or the newest + release stored with `exampleSource`. 6. **Tests:** `TestSections` checks the new section implements `Section` and that its keys match its `cfg` tag, with no changes needed. In `internal/config/conf_sections_test.go`, add the section to diff --git a/internal/cli/fetch/service.go b/internal/cli/fetch/service.go index 1a219d7..58054c4 100644 --- a/internal/cli/fetch/service.go +++ b/internal/cli/fetch/service.go @@ -79,7 +79,8 @@ func commandService(opts *cli.Options) *cobra.Command { []string{ time.DateOnly, time.DateTime, }, - "contract updates published since YYYY-MM-DD[ hh:mm:ss]", + "contract updates published since YYYY-MM-DD[ hh:mm:ss] "+ + "(default: each source's newest stored release, or the last day)", ) cmd.Flags().Bool("foreground", false, "run in the foreground") diff --git a/internal/cli/serve/command.go b/internal/cli/serve/command.go index 50ccd7a..570f128 100644 --- a/internal/cli/serve/command.go +++ b/internal/cli/serve/command.go @@ -176,7 +176,8 @@ sources.`, []string{ time.DateOnly, time.DateTime, }, - "contract updates published since YYYY-MM-DD[ hh:mm:ss]", + "contract updates published since YYYY-MM-DD[ hh:mm:ss] "+ + "(default: each source's newest stored release, or the last day)", ) cmd.MarkFlagsMutuallyExclusive("listen", "host") cmd.MarkFlagsMutuallyExclusive("listen", "port") diff --git a/internal/fetcher/fetcher.go b/internal/fetcher/fetcher.go index 436084e..e370388 100644 --- a/internal/fetcher/fetcher.go +++ b/internal/fetcher/fetcher.go @@ -99,6 +99,11 @@ func New(conf *config.JSCSources, opts ...Option) *Fetcher { // 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. +// +// Every worker starts from since. A zero since starts each source from the +// newest release stored for it (see WithStore and startFrom); with no store +// or nothing stored, the source's client picks (Find a Tender's: the last +// day). func (f *Fetcher) Run(ctx context.Context, since time.Time) error { defer f.backgroundCancel() @@ -114,14 +119,22 @@ func (f *Fetcher) Run(ctx context.Context, since time.Time) error { errs []error ) - // 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) { + // start runs one source's worker from where it starts (see startFrom) + // until f.backgroundCtx is cancelled. name is also the Source its + // opportunities are stored with. 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, since time.Time) error, + ) { wg.Add(1) go func() { defer wg.Done() - if err := run(f.backgroundCtx); err != nil { + sourceSince, err := f.startFrom(f.backgroundCtx, logger, name, since) + if err == nil { + err = run(f.backgroundCtx, sourceSince) + } + if err != nil { tagged(logger).Error( "source stopped, shutting down", "source", name, @@ -136,17 +149,17 @@ func (f *Fetcher) Run(ctx context.Context, since time.Time) error { } if f.conf.FindATender.Enabled { - start("FindATender", func(ctx context.Context) error { + start(findATenderSource, func(ctx context.Context, since time.Time) error { return runFindATender(ctx, logger, &f.conf.FindATender, f.save, since) }) } if f.conf.FTSMock.Enabled { - start("FTSMock", func(ctx context.Context) error { + start(ftsMockSource, func(ctx context.Context, since time.Time) error { return runFTSMock(ctx, logger, &f.conf.FTSMock, f.save, since) }) } if f.conf.Sleeper.Enabled { - start("Sleeper", func(ctx context.Context) error { + start("Sleeper", func(ctx context.Context, since time.Time) error { return runSleeper(ctx, logger, &f.conf.Sleeper, since) }) } @@ -158,6 +171,34 @@ func (f *Fetcher) Run(ctx context.Context, since time.Time) error { return errors.Join(errs...) } +// startFrom returns the time source's worker starts from: since if it is +// set, as an override. Otherwise it is the date of the newest release stored +// for source, so a restart carries on where the last run left off, or the +// zero time with no store or nothing stored yet, which leaves the start to +// the source's client. Fetching from the newest release's date fetches it +// again; the store skips it. +func (f *Fetcher) startFrom( + ctx context.Context, logger *slog.Logger, source string, since time.Time, +) (time.Time, error) { + if !since.IsZero() || f.store == nil { + return since, nil + } + + latest, err := f.store.LatestReleaseDate(ctx, source) + if err != nil { + return time.Time{}, err + } + if !latest.IsZero() { + tagged(logger).Info( + "starting from the newest stored release", + "source", source, + "since", latest, + ) + } + + return latest, nil +} + // 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. diff --git a/internal/fetcher/fetcher_test.go b/internal/fetcher/fetcher_test.go index fb407ae..1e7112d 100644 --- a/internal/fetcher/fetcher_test.go +++ b/internal/fetcher/fetcher_test.go @@ -552,3 +552,110 @@ func TestSave(t *testing.T) { t.Errorf("save to a closed store = %v; want a saving error", err) } } + +// updatedFromServer serves emptyPackage and sends the updatedFrom of each +// request to the returned channel. +func updatedFromServer(t *testing.T) (*httptest.Server, chan string) { + t.Helper() + updatedFrom := make(chan string, 10) + srv := httptest.NewServer( + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + select { + case updatedFrom <- r.URL.Query().Get("updatedFrom"): + default: + } + io.WriteString(w, emptyPackage) + }), + ) + t.Cleanup(srv.Close) + return srv, updatedFrom +} + +// Without since, each source starts from its own newest stored release, or +// with nothing stored from the client's default (the last day); since +// overrides both. +func TestRunStartsFromNewestStoredRelease(t *testing.T) { + opportunityStore := openTestStore(t) + srv, _ := ftsServer(t, http.StatusOK, onePackage) + stored, err := fetchers["FindATender"]( + context.Background(), srv.URL, time.Now().Add(-time.Hour), + ) + if err != nil { + t.Fatal(err) + } + _, err = opportunityStore.SaveOpportunities(context.Background(), stored) + if err != nil { + t.Fatal(err) + } + + override := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) + tests := []struct { + name string + since time.Time + wantFAT string + wantMock string + }{ + // "" is the client's default: about a day ago. + {"stored", time.Time{}, "2026-10-01T12:00:00", ""}, + {"override", override, "2026-09-01T00:00:00", "2026-09-01T00:00:00"}, + } + for _, tt := range tests { + fat, fatFrom := updatedFromServer(t) + mock, mockFrom := updatedFromServer(t) + 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, tt.since) + }() + + for source, got := range map[string]chan string{ + "FindATender": fatFrom, + "FTSMock": mockFrom, + } { + want := tt.wantFAT + if source == "FTSMock" { + want = tt.wantMock + } + select { + case from := <-got: + if want == "" { + checkAboutADayAgo(t, from) + } else if from != want { + t.Errorf( + "%s: %s updatedFrom = %s; want %s", + tt.name, source, from, want, + ) + } + case <-time.After(5 * time.Second): + t.Errorf("%s: %s was not polled", tt.name, source) + } + } + cancel() + <-done + } +} + +// checkAboutADayAgo fails the test unless updatedFrom, in local time as the +// client sends it, is about a day ago: within an hour, as a calendar day +// across a clock change isn't 24 hours. +func checkAboutADayAgo(t *testing.T, updatedFrom string) { + t.Helper() + from, err := time.ParseInLocation( + "2006-01-02T15:04:05", updatedFrom, time.Local, + ) + if err != nil { + t.Fatal(err) + } + gap := time.Since(from) - 24*time.Hour + if gap < -time.Hour || gap > time.Hour { + t.Errorf("updatedFrom = %s; want about a day ago", updatedFrom) + } +} diff --git a/internal/fetcher/worker_findatender.go b/internal/fetcher/worker_findatender.go index f1ae17b..6421720 100644 --- a/internal/fetcher/worker_findatender.go +++ b/internal/fetcher/worker_findatender.go @@ -12,6 +12,10 @@ import ( "git.avsdev.uk/Go/job-search/internal/util" ) +// findATenderSource is the Source of the opportunities Find a Tender's +// converter makes, and so of the ones the Find a Tender source fetches. +const findATenderSource = "FindATender" + // runFindATender polls Find a Tender every configured interval until ctx // is cancelled, passing what each run fetches to save. func runFindATender( @@ -27,7 +31,7 @@ func runFindATender( worker := util.Worker{ Logger: tagged(logger), - Name: "FindATender", + Name: findATenderSource, Interval: conf.Interval, Task: func(ctx context.Context, since time.Time) error { opportunities, err := fetchFindATender(ctx, logger, conf, since) diff --git a/internal/store/queries.go b/internal/store/queries.go index 3c59243..ff32c2a 100644 --- a/internal/store/queries.go +++ b/internal/store/queries.go @@ -23,6 +23,16 @@ package store const selectOpportunityLatestReleaseDate = ` SELECT latest_release_date FROM opportunities WHERE uuid = ?` +// selectSourceLatestReleaseDate reads the newest latest_release_date of one +// source's opportunities. ORDER BY rather than MAX keeps the column's DATETIME +// type, so the driver reads it as a time; it walks the latest_release_date +// index. +const selectSourceLatestReleaseDate = ` +SELECT latest_release_date FROM opportunities +WHERE source = ? +ORDER BY latest_release_date DESC +LIMIT 1` + // upsertOpportunity inserts or updates a opportunity. Named parameters // (types.Opportunity). // diff --git a/internal/store/queries_test.go b/internal/store/queries_test.go index e893fa4..ab4421b 100644 --- a/internal/store/queries_test.go +++ b/internal/store/queries_test.go @@ -33,6 +33,9 @@ var checks = map[string]queryCheck{ "selectOpportunityLatestReleaseDate": { sql: selectOpportunityLatestReleaseDate, }, + "selectSourceLatestReleaseDate": { + sql: selectSourceLatestReleaseDate, + }, "upsertOpportunity": {sql: upsertOpportunity, named: true}, "selectOpportunityByUUID": {sql: selectOpportunityByUUID}, "listOpportunitiesSelect": { diff --git a/internal/store/store.go b/internal/store/store.go index 9571c5c..eb8f717 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -716,6 +716,26 @@ func (s *Store) ListReleases( return releases, nil } +// LatestReleaseDate returns the date of the newest release stored for the +// opportunities from source, or the zero time if none are stored, so a +// fetcher can carry on from where it left off. +func (s *Store) LatestReleaseDate( + ctx context.Context, source string, +) (time.Time, error) { + var latest time.Time + err := s.db.GetContext(ctx, &latest, selectSourceLatestReleaseDate, source) + if errors.Is(err, sql.ErrNoRows) { + return time.Time{}, nil + } + if err != nil { + return time.Time{}, fmt.Errorf( + "reading the latest release date of %s: %w", source, err, + ) + } + + return latest, nil +} + // ReleaseExists reports whether a release is already stored, so a fetcher // can skip converting it. func (s *Store) ReleaseExists( diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 4396d61..99483a4 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -837,3 +837,37 @@ func TestWonAndLostAreExclusive(t *testing.T) { } } } + +// LatestReleaseDate is the newest release of one source's opportunities only, +// and the zero time for a source with none stored. +func TestLatestReleaseDate(t *testing.T) { + s := openStore(t) + ctx := context.Background() + + latest, err := s.LatestReleaseDate(ctx, "FindATender") + if err != nil || !latest.IsZero() { + t.Fatalf("empty store: %v, %v; want the zero time", latest, err) + } + + mustSave(t, s, opportunity("ocds-1", "Older", 1, "r1")) + mustSave(t, s, opportunity("ocds-2", "Newer", 5, "r1")) + other := opportunity("ocds-3", "Another source's", 9, "r1") + other.Source = "FTSMock" + mustSave(t, s, other) + + for source, uuid := range map[string]string{ + "FindATender": "ocds-2", + "FTSMock": "ocds-3", + } { + want := mustGet(t, s, uuid).LatestReleaseDate + latest, err := s.LatestReleaseDate(ctx, source) + if err != nil || !latest.Equal(want) { + t.Errorf("%s: %v, %v; want %v", source, latest, err, want) + } + } + + latest, err = s.LatestReleaseDate(ctx, "Nowhere") + if err != nil || !latest.IsZero() { + t.Errorf("unknown source: %v, %v; want the zero time", latest, err) + } +}