job-search/internal/findatender/api.go

package findatender

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"math"
	"regexp"
	"strconv"
	"strings"
	"time"

	resty "resty.dev/v3"

	"git.avsdev.uk/Go/job-search/internal/findatender/ocds"
	"git.avsdev.uk/Go/job-search/internal/logging"
)

// apiTimeLayout is the layout of updatedFrom and updatedTo: wall-clock time
// with no offset, which the API reads as UK time.
const apiTimeLayout = "2006-01-02T15:04:05"

// Errors returned when fetching from the API.
var (
	// ErrNoNext means a release package has no next page link.
	ErrNoNext = errors.New("findatender: no next link")
	// ErrNoReleases means the API returned no releases.
	ErrNoReleases = errors.New("findatender: no releases")
	// ErrRateLimitExceeded means the API refused a request with 429 Too Many
	// Requests, and waiting for it was not allowed or did not help (see
	// RequestOptions.RateLimitRetries).
	ErrRateLimitExceeded = errors.New("findatender: rate limit has been exceeded")
	// ErrFetchCancelled means the fetch's context was cancelled while waiting
	// out a rate limit.
	ErrFetchCancelled = errors.New("findatender: fetch cancelled")
	// ErrUnexpectedStatus wraps an API response status other than 200 OK.
	ErrUnexpectedStatus = errors.New("findatender: unexpected response status")
)

// RequestOptions selects the releases a fetch asks for.
type RequestOptions struct {
	// Since asks for releases updated from this time. Zero means 24 hours ago.
	Since time.Time
	// Until asks for releases updated up to this time. Zero means no bound
	// (the API's: now).
	Until time.Time
	// Limit is the number of releases per page. Zero means 100.
	Limit int
	// RateLimitRetries is how many times a request refused with 429 Too Many
	// Requests is retried, each after waiting as the response's Retry-After
	// asks, before failing with ErrRateLimitExceeded. 0 fails at the first.
	RateLimitRetries int
}

// ---------- Utility methods for handling API irregularities ----------

// safeReleasePackage is ocds.ReleasePackage with the releases left undecoded,
// so safeUnpackReleases can repair or skip each one.
type safeReleasePackage struct {
	URI               string                        `json:"uri,omitempty"`
	Version           string                        `json:"version,omitempty"`
	Extensions        []string                      `json:"extensions,omitempty"`
	PublishedDate     *string                       `json:"publishedDate,omitempty"`
	Releases          []json.RawMessage             `json:"releases,omitempty"`
	Publisher         *ocds.ReleasePackagePublisher `json:"publisher,omitempty"`
	License           string                        `json:"license,omitempty"`
	PublicationPolicy string                        `json:"publicationPolicy,omitempty"`
	Links             *ocds.ReleasePackageLinks     `json:"links,omitempty"`
}

// dateOnly matches a calendar date with no time of day: "2006-10-02".
var dateOnly = regexp.MustCompile(`^\d{4}-\d{2}-\d{2}$`)

// unmarshalLenient decodes raw json into target like json.Unmarshal, after
// repairing values that it cannot decode (see repairJSON).
func unmarshalLenient(raw json.RawMessage, target any) error {
	fixed, err := repairJSON(raw)
	if err != nil {
		return err
	}
	return json.Unmarshal(fixed, target)
}

// repairJSON returns a copy of raw with the values json.Unmarshal cannot
// decode repaired:
//
//   - every date field holding a date-only value ("2006-10-02") is rewritten
//     as midnight UTC ("2006-10-02T00:00:00Z"), and every date field holding
//     an empty string is set to null. Values that already have a time, values
//     in other fields, and dates that appear inside free text are left alone.
//     A date field is one whose name is "date" or starts with "date",
//     contains "Date", or ends with "Deadline".
//   - every number too large for a float64 is replaced by math.MaxInt64 (or
//     math.MinInt64 if negative). OCDS writes "no limit" as 1e9999, as JSON
//     has no infinity; math.MaxInt64 still fits the int and float64 fields
//     it is found in, and is marshalled again where +Inf would not be.
//
// If nothing needs repairing, raw is returned unchanged.
func repairJSON(raw json.RawMessage) (json.RawMessage, error) {
	decoder := json.NewDecoder(bytes.NewReader(raw))
	decoder.UseNumber() // keep numbers exactly as written, not as float64

	var doc any
	if err := decoder.Decode(&doc); err != nil {
		return nil, fmt.Errorf("repairing json: %w", err)
	}

	if _, err := decoder.Token(); !errors.Is(err, io.EOF) {
		return nil, errors.New(
			"findatender: repairing json: unexpected data after the JSON value",
		)
	}

	repaired, changed := recursiveRepair("", doc)
	if !changed {
		return raw, nil
	}

	return json.Marshal(repaired)
}

// recursiveRepair repairs value, found in the field named key ("" for an
// array element or the document itself), and its contents (see repairJSON).
// Maps and slices are repaired in place; the repaired value is returned, and
// whether anything changed.
func recursiveRepair(key string, value any) (any, bool) {
	switch typed := value.(type) {
	case map[string]any:
		changed := false
		for fieldKey, fieldValue := range typed {
			repaired, fieldChanged := recursiveRepair(fieldKey, fieldValue)
			if fieldChanged {
				typed[fieldKey] = repaired
				changed = true
			}
		}
		return typed, changed
	case []any:
		changed := false
		for i, element := range typed {
			repaired, elementChanged := recursiveRepair("", element)
			if elementChanged {
				typed[i] = repaired
				changed = true
			}
		}
		return typed, changed
	case string:
		return repairDate(key, typed)
	case json.Number:
		return repairNumber(typed)
	default:
		return value, false
	}
}

// repairDate repairs text if key is a date field (see repairJSON).
func repairDate(key, text string) (any, bool) {
	if !isDateField(key) {
		return text, false
	}
	switch {
	case dateOnly.MatchString(text):
		return text + "T00:00:00Z", true
	case text == "":
		return nil, true
	default:
		return text, false
	}
}

// repairNumber replaces a number too large for a float64 (see repairJSON).
func repairNumber(number json.Number) (any, bool) {
	parsed, err := strconv.ParseFloat(number.String(), 64)
	switch {
	case !errors.Is(err, strconv.ErrRange) || !math.IsInf(parsed, 0):
		return number, false
	case parsed > 0:
		return json.Number(strconv.FormatInt(math.MaxInt64, 10)), true
	default:
		return json.Number(strconv.FormatInt(math.MinInt64, 10)), true
	}
}

// isDateField reports whether a field holds a date, by its name (see
// repairJSON).
func isDateField(name string) bool {
	return strings.HasPrefix(name, "date") ||
		strings.Contains(name, "Date") ||
		strings.HasSuffix(name, "Deadline")
}

// releaseIDs reads just the ocid and the release id from a release, without
// decoding the rest of it. It reads the top-level fields in order and stops
// as soon as it has both, so an error later in the release (invalid JSON or a
// value that wouldn't fit the release struct) doesn't matter. Fields before
// them only need to be valid JSON.
func releaseIDs(raw json.RawMessage) (ocid, id string, err error) {
	decoder := json.NewDecoder(bytes.NewReader(raw))
	decoder.UseNumber() // a numeric id keeps its exact digits

	token, err := decoder.Token()
	if err != nil {
		return "", "", fmt.Errorf("reading release: %w", err)
	}
	if token != json.Delim('{') {
		return "", "", errors.New("findatender: reading release: not a JSON object")
	}

	var haveOCID, haveID bool
	for decoder.More() {
		token, err := decoder.Token()
		if err != nil {
			return ocid, id, fmt.Errorf("reading release: %w", err)
		}
		switch key, _ := token.(string); key {
		case "ocid":
			if err := decoder.Decode(&ocid); err != nil {
				return ocid, id, fmt.Errorf("reading release ocid: %w", err)
			}
			haveOCID = true
		case "id":
			if id, err = decodeReleaseID(decoder); err != nil {
				return ocid, id, err
			}
			haveID = true
		default:
			var skip json.RawMessage // must be valid JSON, but is otherwise ignored
			if err := decoder.Decode(&skip); err != nil {
				return ocid, id, fmt.Errorf("reading release field %q: %w", key, err)
			}
		}
		if haveOCID && haveID {
			return ocid, id, nil
		}
	}

	return ocid, id, missingIDsError(haveOCID, haveID)
}

// decodeReleaseID decodes a release id. OCDS ids are strings, but a number is
// tolerated and kept as written.
func decodeReleaseID(decoder *json.Decoder) (string, error) {
	var decoded any
	if err := decoder.Decode(&decoded); err != nil {
		return "", fmt.Errorf("reading release id: %w", err)
	}
	switch typed := decoded.(type) {
	case string:
		return typed, nil
	case json.Number:
		return typed.String(), nil
	default:
		return "", fmt.Errorf("findatender: reading release id: unexpected %T", typed)
	}
}

// missingIDsError reports which of the ocid and release id were not found.
func missingIDsError(haveOCID, haveID bool) error {
	switch {
	case !haveOCID && !haveID:
		return errors.New("findatender: reading release: no ocid or id")
	case !haveOCID:
		return errors.New("findatender: reading release: no ocid")
	default:
		return errors.New("findatender: reading release: no id")
	}
}

// safeUnpackReleases unpacks an API response into an OCDS ReleasePackage,
// handling API irregularities (malformed dates, 1e9999 for no limit, missing
// fields, invalid releases, etc).
//
// This method takes ownership of the response body and closes it when done.
func safeUnpackReleases(
	logger *slog.Logger, response *resty.Response,
) (*ocds.ReleasePackage, error) {
	var releasePackage safeReleasePackage

	// Take ownership of the response body.
	defer response.Body.Close() //nolint:errcheck // read only

	decoder := json.NewDecoder(response.Body)
	if err := decoder.Decode(&releasePackage); err != nil {
		return nil, err
	}

	out := &ocds.ReleasePackage{
		URI:               releasePackage.URI,
		Version:           releasePackage.Version,
		Extensions:        releasePackage.Extensions,
		Publisher:         releasePackage.Publisher,
		License:           releasePackage.License,
		PublicationPolicy: releasePackage.PublicationPolicy,
		Links:             releasePackage.Links,
	}

	// Correct PublishedDate of "" and "2006-01-02"
	if releasePackage.PublishedDate != nil &&
		*releasePackage.PublishedDate != "" {
		published, err := time.Parse(time.RFC3339, *releasePackage.PublishedDate)
		if err != nil {
			published, err = time.Parse(time.DateOnly, *releasePackage.PublishedDate)
		}
		if err == nil {
			out.PublishedDate = &published
		}
	}

	for _, rawRelease := range releasePackage.Releases {
		var release ocds.Release
		err := unmarshalLenient(rawRelease, &release)
		if err != nil {
			ocid, releaseID, _ := releaseIDs(rawRelease)
			logger.Warn(
				"skipping unreadable release",
				"ocid", ocid,
				"release_id", releaseID,
				"error", err,
			)
			logger.Debug(
				"raw unreadable release",
				"ocid", ocid,
				"release_id", releaseID,
				"release_raw", rawRelease,
			)
			continue
		}
		out.Releases = append(out.Releases, release)
	}

	return out, nil
}

// ---------- API interaction methods ----------

// rateSleep sleeps the fetch worker until either the context is cancelled
// (shutdown), returning ctx.Err(), or the duration has expired. The duration
// is usually provided by the API; a zero duration returns immediately.
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()
	}
}

// closeBody closes res's body if it has one, avoiding a double close.
func closeBody(res *resty.Response) {
	if res.Body != nil {
		res.Body.Close() //nolint:errcheck,gosec // read only
	}
}

// fetchURI performs a single GET of uri, using findATender's context, and
// traces it when debug logging is on (logging.IsDebugEnabled). Unlike the
// method of the same name it does not check the response status.
func fetchURI(findATender *FindATender, uri string) (*resty.Response, error) {
	withTrace := logging.IsDebugEnabled()
	logger := findATender.logger()

	logger.Info("fetching", "uri", uri)

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

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

	if withTrace {
		logTrace(logger, res)
	}

	return res, err
}

// logTrace logs res's trace, status and body at debug level. On a terminal
// the trace and body are printed as they are, as a field would escape them
// into one line; elsewhere they are fields, the trace as JSON for the JSON
// (file) logger.
func logTrace(logger *slog.Logger, res *resty.Response) {
	if logging.WritesToTerminal(logger.Handler()) {
		logger.Debug("fetched", "status", res.Status())
		logging.PrintToTerminal(res.Request.TraceInfo().String())
		logging.PrintToTerminal(res.String())
		return
	}

	trace := res.Request.TraceInfo().String()
	if logging.IsJSONHandler(logger.Handler()) {
		trace = res.Request.TraceInfo().JSON()
	}
	if len(res.String()) <= 1024 {
		logger.Debug(
			"fetched",
			"trace", trace,
			"status", res.Status(),
			"response", res.String(),
		)
	} else {
		logger.Debug(
			"fetched",
			"trace", trace,
			"status", res.Status(),
			"response_length", len(res.String()),
		)
	}
}

// fetchURI performs a fetch against the API. A response of 429 Too Many
// Requests is waited out for as long as its Retry-After asks and the request
// retried, up to rateLimitRetries times, like a good client; one more fails
// with ErrRateLimitExceeded.
func (ft *FindATender) fetchURI(
	uri string, rateLimitRetries int,
) (*resty.Response, error) {
	ctx := ft.ctx
	if ctx == nil {
		ctx = context.Background()
	}

	res, err := fetchURI(ft, uri)
	for retry := 1; err == nil && res.StatusCode() == 429; retry++ {
		closeBody(res)

		if retry > rateLimitRetries {
			return nil, ErrRateLimitExceeded
		}

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

		retryAfter := time.Duration(retryAfterSeconds) * time.Second
		ft.logger().Warn(
			"api rate limit hit, waiting as asked",
			"retry_after", retryAfter,
			"retry", retry,
			"retries", rateLimitRetries,
		)
		if rateSleep(ctx, retryAfter) != nil {
			return nil, ErrFetchCancelled
		}

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

	if res.StatusCode() != 200 {
		closeBody(res)
		return nil, fmt.Errorf("%w: %s", ErrUnexpectedStatus, res.Status())
	}

	return res, nil
}

// fetchWithOpts performs a fetch against the API with some sanitised default
// options (see RequestOptions).
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(apiTimeLayout),
		opts.Limit,
	)
	if !opts.Until.IsZero() {
		uri += "&updatedTo=" + opts.Until.Format(apiTimeLayout)
	}

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

// fetchNext fetches the page pkg's "Next" link points to. It returns
// ErrNoNext if there is no "Next" link; the caller can decide whether asking
// for it was an error.
func (ft *FindATender) fetchNext(
	releasePackage *ocds.ReleasePackage, rateLimitRetries int,
) (*resty.Response, error) {
	if releasePackage.Links == nil || releasePackage.Links.Next == "" {
		return nil, ErrNoNext
	}
	return ft.fetchURI(releasePackage.Links.Next, rateLimitRetries)
}

// fetchOnce makes one call on the API. It does not follow any "Next" links.
func (ft *FindATender) fetchOnce(opts RequestOptions) ([]ocds.Release, error) {
	res, err := ft.fetchWithOpts(opts)
	if err != nil {
		return nil, err
	}

	releasePackage, err := safeUnpackReleases(ft.logger(), res)
	if err != nil {
		return nil, err
	}

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

	return releasePackage.Releases, nil
}

// fetchAll makes a call on the API. If there are any "Next" links, it calls
// them in succession until the end has been reached. If a page fails, the
// releases of the pages before it are returned with the error, so the caller
// can keep them.
func (ft *FindATender) fetchAll(opts RequestOptions) ([]ocds.Release, error) {
	var releases []ocds.Release

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

	releasePackage, err := safeUnpackReleases(ft.logger(), res)
	if err != nil {
		return nil, err
	}
	releases = append(releases, releasePackage.Releases...)

	// Next links already followed, so a page that links back to an earlier
	// one cannot loop forever.
	seen := make(map[string]bool)
	for {
		if releasePackage.Links != nil && releasePackage.Links.Next != "" {
			if seen[releasePackage.Links.Next] {
				ft.logger().Warn(
					"stopping at repeated next link",
					"link", releasePackage.Links.Next,
				)
				break
			}
			seen[releasePackage.Links.Next] = true
		}

		res, err := ft.fetchNext(releasePackage, opts.RateLimitRetries)
		if errors.Is(err, ErrNoNext) {
			break
		}
		if err != nil {
			return releases, err
		}

		releasePackage, err = safeUnpackReleases(ft.logger(), res)
		if err != nil {
			return releases, err
		}
		releases = append(releases, releasePackage.Releases...)
	}

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

	return releases, nil
}