job-search/internal/fetcher/backfill_test.go

package fetcher

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"git.avsdev.uk/Go/job-search/internal/config"
	"git.avsdev.uk/Go/job-search/internal/types"
)

var errTestFetch = errors.New("test fetch failed")

// span describes the window from since to until, for comparing windows.
func span(since, until time.Time) string {
	return since.Format(time.RFC3339) + "/" + until.Format(time.RFC3339)
}

// recorder is a fake source for backfill: it records the windows fetched
// and the saves, in order, in calls.
type recorder struct {
	mu      sync.Mutex
	calls   []string
	windows []string
	saved   int
	// fail returns the error for the nth fetch (from 1), if any.
	fail func(n int) error
}

func (r *recorder) fetch(
	ctx context.Context, since, until time.Time,
) ([]*types.Opportunity, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.windows = append(r.windows, span(since, until))
	r.calls = append(r.calls, "fetch")
	var err error
	if r.fail != nil {
		err = r.fail(len(r.windows))
	}
	opportunity := &types.Opportunity{
		UUID: "ocds-1",
	}
	return []*types.Opportunity{opportunity}, err
}

func (r *recorder) save(
	ctx context.Context, logger *slog.Logger, opportunities []*types.Opportunity,
) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.saved += len(opportunities)
	r.calls = append(r.calls, "save")
	return nil
}

func TestBackfillWindows(t *testing.T) {
	since := time.Now().Add(-20 * 24 * time.Hour)
	r := &recorder{}

	got := backfill(
		context.Background(),
		slog.Default(),
		"Test",
		time.Millisecond,
		r.fetch,
		r.save,
		since,
	)

	second := since.Add(backfillChunk)
	third := since.Add(2 * backfillChunk)
	want := []string{span(since, second), span(second, third)}
	if !slices.Equal(r.windows, want) {
		t.Errorf("windows = %v; want %v", r.windows, want)
	}
	if !got.Equal(third) {
		t.Errorf("backfill = %v; want the end of the last window", got)
	}
	// Each window is saved before the next is fetched.
	if calls := strings.Join(r.calls, ","); calls != "fetch,save,fetch,save" {
		t.Errorf("calls = %s", calls)
	}
}

// Nothing is backfilled for a zero since (the client picks) or one within a
// window of now: the worker fetches it.
func TestBackfillNothingToDo(t *testing.T) {
	for name, since := range map[string]time.Time{
		"zero":   {},
		"recent": time.Now().Add(-3 * 24 * time.Hour),
		"future": time.Now().Add(time.Hour),
	} {
		t.Run(name, func(t *testing.T) {
			r := &recorder{}
			got := backfill(
				context.Background(),
				slog.Default(),
				"Test",
				time.Millisecond,
				r.fetch,
				r.save,
				since,
			)
			if !got.Equal(since) || len(r.windows) != 0 {
				t.Errorf(
					"backfill = %v after %d fetches; want %v and none",
					got, len(r.windows), since,
				)
			}
		})
	}
}

// A failed window is saved as far as it got and fetched again from its
// start; finding nothing moves on like a success.
func TestBackfillRetriesFailedWindow(t *testing.T) {
	since := time.Now().Add(-15 * 24 * time.Hour)
	r := &recorder{
		fail: func(n int) error {
			switch n {
			case 1, 2:
				return errTestFetch
			case 3:
				return ErrNoNewReleases
			default:
				return nil
			}
		},
	}

	got := backfill(
		context.Background(),
		slog.Default(),
		"Test",
		time.Millisecond,
		r.fetch,
		r.save,
		since,
	)

	second := since.Add(backfillChunk)
	third := since.Add(2 * backfillChunk)
	first := span(since, second)
	want := []string{first, first, first, span(second, third)}
	if !slices.Equal(r.windows, want) {
		t.Errorf("windows = %v; want %v", r.windows, want)
	}
	if r.saved != 4 {
		t.Errorf("saved %d opportunities; want 4 (failed windows too)", r.saved)
	}
	if !got.Equal(third) {
		t.Errorf("backfill = %v; want %v", got, third)
	}
}

// Cancelling stops the backfill during its wait, returning the start of the
// window it stopped in.
func TestBackfillCancelled(t *testing.T) {
	since := time.Now().Add(-15 * 24 * time.Hour)
	ctx, cancel := context.WithCancel(context.Background())
	r := &recorder{
		fail: func(n int) error {
			cancel()
			return errTestFetch
		},
	}

	start := time.Now()
	got := backfill(
		ctx, slog.Default(), "Test", time.Hour, r.fetch, r.save, since,
	)

	if time.Since(start) > 5*time.Second {
		t.Error("cancelling did not stop the wait")
	}
	if !got.Equal(since) || len(r.windows) != 1 {
		t.Errorf(
			"backfill = %v after %d fetches; want %v after 1",
			got, len(r.windows), since,
		)
	}
}

func TestSaveFetched(t *testing.T) {
	r := &recorder{}
	partial := []*types.Opportunity{
		{
			UUID: "ocds-1",
		},
	}

	err := saveFetched(
		context.Background(), slog.Default(), r.save, partial, errTestFetch,
	)
	if !errors.Is(err, errTestFetch) || r.saved != 1 {
		t.Errorf(
			"partial: err = %v, saved %d; want the fetch error, 1", err, r.saved,
		)
	}

	err = saveFetched(
		context.Background(), slog.Default(), r.save, nil, ErrNoNewReleases,
	)
	if !errors.Is(err, ErrNoNewReleases) || r.saved != 1 {
		t.Errorf(
			"nothing: err = %v, saved %d; want ErrNoNewReleases, 1", err, r.saved,
		)
	}

	failing := func(context.Context, *slog.Logger, []*types.Opportunity) error {
		return errTestFetch
	}
	err = saveFetched(context.Background(), slog.Default(), failing, partial, nil)
	if !errors.Is(err, errTestFetch) {
		t.Errorf("failed save: err = %v; want the save's error", err)
	}
}

// Run backfills each Find a Tender source in windows bounded on both sides,
// then its worker fetches the rest with no upper bound.
func TestRunBackfills(t *testing.T) {
	for _, source := range []string{"FindATender", "FTSMock"} {
		t.Run(source, func(t *testing.T) {
			var (
				mu    sync.Mutex
				until []string
			)
			srv := httptest.NewServer(
				http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
					mu.Lock()
					until = append(until, r.URL.Query().Get("updatedTo"))
					mu.Unlock()
					io.WriteString(w, emptyPackage)
				}),
			)
			t.Cleanup(srv.Close)

			sources := &config.JSCSources{}
			if source == "FindATender" {
				sources.FindATender.Enabled = true
				sources.FindATender.Interval = time.Hour
				sources.FindATender.APIRoot = srv.URL
			} else {
				sources.FTSMock.Enabled = true
				sources.FTSMock.Interval = time.Hour
				sources.FTSMock.APIRoot = srv.URL
			}

			since := time.Now().UTC().Add(-15 * 24 * time.Hour)
			ctx, cancel := context.WithCancel(context.Background())
			done := make(chan error, 1)
			go func() {
				done <- New(sources).Run(ctx, since)
			}()

			deadline := time.Now().Add(5 * time.Second)
			for {
				mu.Lock()
				n := len(until)
				mu.Unlock()
				if n >= 3 || time.Now().After(deadline) {
					break
				}
				time.Sleep(5 * time.Millisecond)
			}
			cancel()
			if err := <-done; err != nil {
				t.Fatalf("Run = %v", err)
			}

			const layout = "2006-01-02T15:04:05"
			want := []string{
				since.Add(backfillChunk).Format(layout),
				since.Add(2 * backfillChunk).Format(layout),
				"",
			}
			mu.Lock()
			defer mu.Unlock()
			if !slices.Equal(until, want) {
				t.Errorf("updatedTo of each request = %q; want %q", until, want)
			}
		})
	}
}