job-search/internal/fetcher/fetcher.go

package fetcher

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sync"
	"time"

	"git.avsdev.uk/Go/job-search/internal/config"
	"git.avsdev.uk/Go/job-search/internal/events"
	"git.avsdev.uk/Go/job-search/internal/skiprules"
	"git.avsdev.uk/Go/job-search/internal/store"
	"git.avsdev.uk/Go/job-search/internal/types"
	"git.avsdev.uk/Go/job-search/internal/util"
)

// ErrNoNewReleases is returned by a source's task when there was nothing new
// to fetch; it wraps util.ErrWorkerNoWork.
var ErrNoNewReleases = fmt.Errorf("%w: no new releases", util.ErrWorkerNoWork)

// 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(f *Fetcher) {
		f.log = logger
	}
}

// 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
	}
}

// WithEvents makes the workers publish an events.OpportunitiesUpdated event
// to bus whenever a save stores new opportunities or new releases, and an
// events.ClassificationsUpdated event when it changes the classification
// codes in use. It needs WithStore: without a store nothing is saved, so
// nothing is published.
func WithEvents(bus *events.Bus) Option {
	return func(f *Fetcher) {
		f.events = bus
	}
}

// WithSkipRules makes the workers mark the new opportunities that match the
// rules in rulesFile Skipped as they save them (see Fetcher.save). The file
// is checked for changes before each save.
func WithSkipRules(rulesFile *skiprules.File) Option {
	return func(f *Fetcher) {
		f.skipRules = rulesFile
	}
}

// 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.
func (f *Fetcher) logger() *slog.Logger {
	if f.log == nil {
		return slog.Default()
	}
	return f.log
}

// tagged returns logger tagged with the fetcher's component, for the lines
// Run and its workers log themselves.
func tagged(logger *slog.Logger) *slog.Logger {
	return logger.With("component", "fetcher")
}

// 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
	store   *store.Store // nil: nothing is persisted
	events  *events.Bus  // nil: nothing is published
	started time.Time

	// skipRules mark new opportunities Skipped; nil: none are.
	skipRules *skiprules.File

	// 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. 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()

	// Cancelling ctx stops the workers, through the background context.
	stopAfter := context.AfterFunc(ctx, f.backgroundCancel)
	defer stopAfter()

	logger := f.logger()

	var (
		wg   sync.WaitGroup
		mu   sync.Mutex
		errs []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()
			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,
					"error", err,
				)
				mu.Lock()
				errs = append(errs, fmt.Errorf("%s: %w", name, err))
				mu.Unlock()
				f.backgroundCancel()
			}
		}()
	}

	if f.conf.FindATender.Enabled {
		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(ftsMockSource, func(ctx context.Context, since time.Time) error {
			return runFTSMock(ctx, logger, &f.conf.FTSMock, f.save, since)
		})
	}
	if f.conf.EUSupply.Enabled {
		// A nil *store.Store in the interface would not be nil.
		var lookup tenderStore
		if f.store != nil {
			lookup = f.store
		}
		start(euSupplySource, func(ctx context.Context, since time.Time) error {
			return runEUSupply(
				ctx, logger, &f.conf.EUSupply, lookup, f.save, since,
			)
		})
	}
	if f.conf.Sleeper.Enabled {
		start("Sleeper", func(ctx context.Context, since time.Time) 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...)
}

// 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, logs what changed,
// and publishes it (see WithEvents). 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.
//
// Opportunities that match the skip rules (see WithSkipRules) are marked
// Skipped first. The store keeps the marks of an opportunity it already
// has, so only new ones are skipped.
func (f *Fetcher) save(
	ctx context.Context,
	logger *slog.Logger,
	opportunities []*types.Opportunity,
) error {
	if f.store == nil {
		return nil
	}

	filters := f.markSkipped(opportunities)

	result, err := f.store.SaveOpportunities(ctx, opportunities)
	if err != nil {
		return fmt.Errorf("saving opportunities: %w", err)
	}

	skipped := 0
	for i, saved := range result.OpportunityResults {
		if !saved.NewOpportunity || filters[i] == "" {
			continue
		}
		skipped++
		tagged(logger).Debug(
			"skipped new opportunity",
			"uuid", opportunities[i].UUID,
			"filter", filters[i],
		)
	}

	tagged(logger).Info(
		"saved opportunities",
		"new_opportunities", result.NewOpportunities,
		"skipped_opportunities", skipped,
		"updated_opportunities", result.UpdatedOpportunities,
		"new_releases", result.NewReleases,
	)

	f.publishClassificationChanges(result)
	f.publishUpdates(opportunities, result)

	return nil
}

// markSkipped marks the opportunities that match the skip rules Skipped,
// and returns the comment of the filter each matched, indexed like
// opportunities: "" for one that matched none (every one, with no skip
// rules).
func (f *Fetcher) markSkipped(opportunities []*types.Opportunity) []string {
	filters := make([]string, len(opportunities))
	if f.skipRules == nil {
		return filters
	}

	rules := f.skipRules.Rules()
	for i, opportunity := range opportunities {
		comment, matched := rules.Match(opportunity)
		if !matched {
			continue
		}
		opportunity.Marks.Skipped = true
		// A filter's comment may be empty; it still names one here.
		filters[i] = cmp.Or(comment, "(no comment)")
	}

	return filters
}

// publishClassificationChanges publishes an events.ClassificationsUpdated
// event for the classification codes a save, with result, brought into or
// out of use, if it changed any and a bus was given WithEvents. It's
// published before the save's events.OpportunitiesUpdated, so a client
// knows a new code before an opportunity that has it.
func (f *Fetcher) publishClassificationChanges(
	result store.SaveMultipleResult,
) {
	if f.events == nil {
		return
	}
	if len(result.AddedClassifications) == 0 &&
		len(result.RemovedClassificationIDs) == 0 {
		return
	}

	payload := events.ClassificationsUpdatedPayload{
		Added:   result.AddedClassifications,
		Removed: result.RemovedClassificationIDs,
	}
	if payload.Added == nil {
		payload.Added = []types.ClassificationCode{}
	}
	if payload.Removed == nil {
		payload.Removed = []int64{}
	}

	f.events.Publish(events.Event{
		Type:    events.ClassificationsUpdated,
		Payload: payload,
	})
}

// publishUpdates publishes an events.OpportunitiesUpdated event for the
// opportunities that result, from saving them, shows were new or gained new
// releases, if there were any and a bus was given WithEvents. An opportunity
// whose row was only rewritten, such as one fetched again from the newest
// stored release, isn't counted as updated. Over
// events.MaxListedOpportunities, they are only counted (see
// events.OpportunitiesUpdatedPayload).
func (f *Fetcher) publishUpdates(
	opportunities []*types.Opportunity, result store.SaveMultipleResult,
) {
	if f.events == nil {
		return
	}

	payload := events.OpportunitiesUpdatedPayload{
		New:     []string{},
		Updated: []string{},
		Batch:   result.ChangeBatch,
	}
	for i, saved := range result.OpportunityResults {
		switch {
		case saved.NewOpportunity:
			payload.New = append(payload.New, opportunities[i].UUID)
		case saved.NewReleases > 0:
			payload.Updated = append(payload.Updated, opportunities[i].UUID)
		}
	}
	payload.Summary = events.OpportunitiesSummary{
		New:     len(payload.New),
		Updated: len(payload.Updated),
	}
	if payload.Summary.New == 0 && payload.Summary.Updated == 0 {
		return
	}
	if len(payload.New)+len(payload.Updated) > events.MaxListedOpportunities {
		payload.New = []string{}
		payload.Updated = []string{}
		payload.Large = true
	}

	f.events.Publish(events.Event{
		Type:    events.OpportunitiesUpdated,
		Payload: payload,
	})
}