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.
This commit is contained in:
1 parent
3c4fde8dfd
commit
284a589627
10 files changed
+256
-22
No files matched your search
@@ -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")
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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).
|
||||
//
|
||||
|
||||
@@ -33,6 +33,9 @@ var checks = map[string]queryCheck{
|
||||
"selectOpportunityLatestReleaseDate": {
|
||||
sql: selectOpportunityLatestReleaseDate,
|
||||
},
|
||||
"selectSourceLatestReleaseDate": {
|
||||
sql: selectSourceLatestReleaseDate,
|
||||
},
|
||||
"upsertOpportunity": {sql: upsertOpportunity, named: true},
|
||||
"selectOpportunityByUUID": {sql: selectOpportunityByUUID},
|
||||
"listOpportunitiesSelect": {
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user