job-search/internal/findATender/api.go

package findATender

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"strconv"
	"time"

	"git.avsdev.uk/Go/job-search/internal/findATender/ocds"
	"git.avsdev.uk/Go/job-search/internal/logging"
	resty "resty.dev/v3"
)

var ErrNoNext = errors.New("no next link")
var ErrNoReleases = errors.New("no releases")
var ErrRateLimitExceeded = errors.New("rate limit has been exceeded")
var ErrFetchCancelled = errors.New("fetch cancelled")

type RequestOptions struct {
	Since          time.Time
	Limit          int
	AllowRateLimit bool
}

func rateSleep(ctx context.Context, duration time.Duration) error {
	timer := time.NewTimer(duration)
	defer timer.Stop()

	select {
	case <-timer.C:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func fetchURI(ft *FindATender, uri string) (*resty.Response, error) {
	withTrace := ft.config.Logging.LogLevel() <= slog.LevelDebug

	slog.Info("findTender.resty", "request_uri", uri)

	ctx := ft.ctx
	if ctx == nil {
		ctx = context.Background()
	}

	req := ft.httpClient.R()
	req.SetContext(ctx)
	res, err := req.SetTrace(withTrace).Get(uri)

	if withTrace {
		if logging.IsJSONHandler(slog.Default().Handler()) {
			slog.Debug(
				"Resty",
				"trace", res.Request.TraceInfo().JSON(),
				"status", res.Status(),
				"response", res,
			)
		} else {
			slog.Debug("Resty " + res.Request.TraceInfo().String())
			slog.Debug("Resty status: " + res.Status())
			slog.Debug("Resty response: " + res.String())
		}
	}

	return res, err
}

func (ft *FindATender) fetchURI(
	uri string,
	allowRateLimit bool,
) (*resty.Response, error) {
	res, err := fetchURI(ft, uri)
	if err != nil {
		return nil, err
	}

	if ft.ctx == nil {
		copied := *ft
		copied.ctx = context.Background()
		ft = &copied
	}

	if res.StatusCode() == 429 {
		if !allowRateLimit {
			return nil, ErrRateLimitExceeded
		}

		retrySecStr := res.Header().Get("Retry-After")
		retrySec, err := strconv.ParseInt(retrySecStr, 10, 32)
		if err != nil {
			return nil, fmt.Errorf("unknown retry-after value: %w", err)
		}

		slog.Warn(fmt.Sprintf(
			"API rate limit hit, waiting as requested for %d seconds", retrySec,
		))
		if rateSleep(ft.ctx, time.Duration(retrySec)*time.Second) != nil {
			return nil, ErrFetchCancelled
		}

		res, err = fetchURI(ft, uri)
		if err != nil {
			return nil, err
		}
	}

	return res, nil
}

func (ft *FindATender) fetchWithOpts(
	opts RequestOptions,
) (*resty.Response, error) {
	if opts.Since.IsZero() {
		opts.Since = time.Now().AddDate(0, 0, 1)
	}
	if opts.Limit == 0 {
		opts.Limit = 100
	}

	uri := fmt.Sprintf(
		"%s/ocdsReleasePackages?updatedFrom=%s&limit=%d",
		ft.apiRoot,
		opts.Since.Format("2006-01-02T15:04:05"),
		opts.Limit,
	)

	return ft.fetchURI(uri, opts.AllowRateLimit)
}

func (ft *FindATender) fetchNext(
	pkg ocds.ReleasePackage,
	allowRateLimit bool,
) (*resty.Response, error) {
	if pkg.Links == nil || len(pkg.Links.Next) == 0 {
		return nil, ErrNoNext
	}
	return ft.fetchURI(pkg.Links.Next, allowRateLimit)
}

func (ft *FindATender) fetchOnce(opts RequestOptions) ([]ocds.Release, error) {
	res, err := ft.fetchWithOpts(opts)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	var relPkg ocds.ReleasePackage
	decoder := json.NewDecoder(NewOCDSStreamCleaner(res.Body))
	if err := decoder.Decode(&relPkg); err != nil {
		return nil, err
	}

	if len(relPkg.Releases) == 0 {
		return nil, ErrNoReleases
	}

	return relPkg.Releases, nil
}

func (ft *FindATender) fetchAll(opts RequestOptions) ([]ocds.Release, error) {
	var releases []ocds.Release

	res, err := ft.fetchWithOpts(opts)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	var relPkg ocds.ReleasePackage
	decoder := json.NewDecoder(NewOCDSStreamCleaner(res.Body))
	if err := decoder.Decode(&relPkg); err != nil {
		return nil, err
	}
	releases = append(releases, relPkg.Releases...)

	for true {
		res, err := ft.fetchNext(relPkg, opts.AllowRateLimit)
		if err == ErrNoNext {
			break
		}
		if err != nil {
			return nil, err
		}
		defer res.Body.Close()

		decoder := json.NewDecoder(NewOCDSStreamCleaner(res.Body))
		if err := decoder.Decode(&relPkg); err != nil {
			return nil, err
		}
		releases = append(releases, relPkg.Releases...)
	}

	if len(releases) == 0 {
		return nil, ErrNoReleases
	}

	return releases, nil
}