Find a Tender API client updates

- Convert OCDS releases to types.Opportunity (internal/types); FetchOnce
  and FetchAll now return opportunities, built from each ocid's latest
  release
- Include each item's main classification as well as its additional ones
- Return ErrUnexpectedStatus for non-200 responses and
  ErrRateLimitExceeded for a repeated 429; close discarded bodies
- Stop fetchAll on a repeated next link and close each page's body once
  read rather than deferring in the loop
- Default a zero Since to 24 hours ago
- Merge date repair and releaseIDs into api.go
- Filter --planning/--opportunities/--awards on each procedure's current
  stage, using OCDS release tags; --opportunities previously matched a
  nonexistent "opportunity" tag and returned nothing
- Reject conflicting filters before fetching (cobra flag group)

internal/findATender/api.go

@@ -1,12 +1,16 @@
 package findATender
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
+	"regexp"
 	"strconv"
+	"strings"
 	"time"
 
 	"git.avsdev.uk/Go/job-search/internal/findATender/ocds"
@@ -18,6 +22,7 @@ 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")
+var ErrUnexpectedStatus = errors.New("unexpected response status")
 
 type RequestOptions struct {
 	Since          time.Time
@@ -25,6 +30,232 @@ type RequestOptions struct {
 	AllowRateLimit bool
 }
 
+// ---------- Utility methods for handling API irregularities -----------
+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}$`)
+
+// unmarshalLenientDates decodes raw json into v like json.Unmarshal, after
+// repairing date fields that time.Time cannot parse.
+func unmarshalLenientDates(raw json.RawMessage, v any) error {
+	fixed, err := repairDates(raw)
+	if err != nil {
+		return err
+	}
+	return json.Unmarshal(fixed, v)
+}
+
+// repairDates returns a copy of raw in which 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. If nothing needs
+// repairing, raw is returned unchanged.
+//
+// A date field is one whose name is "date" or starts with "date", contains
+// "Date", or ends with "Deadline".
+func repairDates(raw json.RawMessage) (json.RawMessage, error) {
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.UseNumber() // keep numbers exactly as written, not as float64
+
+	var doc any
+	if err := dec.Decode(&doc); err != nil {
+		return nil, fmt.Errorf("repairing dates: %w", err)
+	}
+
+	if _, err := dec.Token(); !errors.Is(err, io.EOF) {
+		return nil, errors.New(
+			"repairing dates: unexpected data after the JSON value",
+		)
+	}
+
+	if !recursiveRepairDates(doc) {
+		return raw, nil
+	}
+
+	return json.Marshal(doc)
+}
+
+// recursiveRepairDates fixes date fields in v in place and reports whether it
+// changed anything.
+func recursiveRepairDates(v any) bool {
+	changed := false
+	switch v := v.(type) {
+	case map[string]any:
+		for key, val := range v {
+			if s, ok := val.(string); ok && isDateField(key) {
+				switch {
+				case dateOnly.MatchString(s):
+					v[key] = s + "T00:00:00Z"
+					changed = true
+				case s == "":
+					v[key] = nil
+					changed = true
+				}
+				continue
+			}
+			if recursiveRepairDates(val) {
+				changed = true
+			}
+		}
+	case []any:
+		for _, el := range v {
+			if recursiveRepairDates(el) {
+				changed = true
+			}
+		}
+	}
+	return changed
+}
+
+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) {
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.UseNumber() // a numeric id keeps its exact digits
+
+	if tok, err := dec.Token(); err != nil {
+		return "", "", fmt.Errorf("reading release: %w", err)
+	} else if tok != json.Delim('{') {
+		return "", "", errors.New("reading release: not a JSON object")
+	}
+
+	var haveOCID, haveID bool
+	for dec.More() {
+		tok, err := dec.Token()
+		if err != nil {
+			return ocid, id, fmt.Errorf("reading release: %w", err)
+		}
+		switch key, _ := tok.(string); key {
+		case "ocid":
+			if err := dec.Decode(&ocid); err != nil {
+				return ocid, id, fmt.Errorf("reading release ocid: %w", err)
+			}
+			haveOCID = true
+		case "id":
+			if id, err = decodeReleaseID(dec); err != nil {
+				return ocid, id, err
+			}
+			haveID = true
+		default:
+			var skip json.RawMessage // must be valid JSON, but is otherwise ignored
+			if err := dec.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(dec *json.Decoder) (string, error) {
+	var v any
+	if err := dec.Decode(&v); err != nil {
+		return "", fmt.Errorf("reading release id: %w", err)
+	}
+	switch v := v.(type) {
+	case string:
+		return v, nil
+	case json.Number:
+		return v.String(), nil
+	default:
+		return "", fmt.Errorf("reading release id: unexpected %T", v)
+	}
+}
+
+// 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("reading release: no ocid or id")
+	case !haveOCID:
+		return errors.New("reading release: no ocid")
+	default:
+		return errors.New("reading release: no id")
+	}
+}
+
+// Unpacks an API response into an OCDS ReleasePackage. Handles API
+// irregularities (malformed dates, missing fields, invalid releases, etc).
+//
+// This method takes ownership of the response body and closes it when done.
+func safeUnpackReleases(response *resty.Response) (*ocds.ReleasePackage, error) {
+	var pkg safeReleasePackage
+
+	// Take ownership of the response body.
+	defer response.Body.Close()
+
+	decoder := json.NewDecoder(response.Body)
+	if err := decoder.Decode(&pkg); err != nil {
+		return nil, err
+	}
+
+	out := &ocds.ReleasePackage{
+		URI:               pkg.URI,
+		Version:           pkg.Version,
+		Extensions:        pkg.Extensions,
+		Publisher:         pkg.Publisher,
+		License:           pkg.License,
+		PublicationPolicy: pkg.PublicationPolicy,
+		Links:             pkg.Links,
+	}
+
+	// Correct PublishedDate of "" and "2006-01-02"
+	if pkg.PublishedDate != nil && len(*pkg.PublishedDate) > 0 {
+		t, err := time.Parse(time.RFC3339, *pkg.PublishedDate)
+		if err != nil {
+			t, err = time.Parse(time.DateOnly, *pkg.PublishedDate)
+		}
+		if err == nil {
+			out.PublishedDate = &t
+		}
+	}
+
+	for _, rawRel := range pkg.Releases {
+		var release ocds.Release
+		err := unmarshalLenientDates(rawRel, &release)
+		if err != nil {
+			ocid, release_id, _ := releaseIDs(rawRel)
+			slog.Warn(fmt.Sprintf(
+				"Skipping release (%s/%s) due to error: %s", ocid, release_id, err,
+			))
+			continue
+		}
+		out.Releases = append(out.Releases, release)
+	}
+
+	return out, nil
+}
+
+// ---------- API interaction methods -----------
+// Sleep the fetch worker until either the context is cancelled (shutdown) or
+// the duration has expired. The duration is usually provided by the API - a
+// nil duration will return immediately.
 func rateSleep(ctx context.Context, duration time.Duration) error {
 	timer := time.NewTimer(duration)
 	defer timer.Stop()
@@ -37,6 +268,14 @@ func rateSleep(ctx context.Context, duration time.Duration) error {
 	}
 }
 
+// Check and close the response body - avoids double close issues
+func closeBody(res *resty.Response) {
+	if res.Body != nil {
+		res.Body.Close()
+	}
+}
+
+// Actually perform a single fetch against the API
 func fetchURI(ft *FindATender, uri string) (*resty.Response, error) {
 	withTrace := ft.config.Logging.LogLevel() <= slog.LevelDebug
 
@@ -69,6 +308,11 @@ func fetchURI(ft *FindATender, uri string) (*resty.Response, error) {
 	return res, err
 }
 
+// Perform a fetch against the API. If the API returns with a rate limit
+// reached response and allowRateLimit is true, then wait and retry like a good
+// client. If allowRateLimit is false, return to the caller informing them of
+// the rate limit exceeded error. If, after waiting, the API still thinks we
+// have exceeded the rate limit, bail out anyway.
 func (ft *FindATender) fetchURI(
 	uri string,
 	allowRateLimit bool,
@@ -85,6 +329,8 @@ func (ft *FindATender) fetchURI(
 	}
 
 	if res.StatusCode() == 429 {
+		closeBody(res)
+
 		if !allowRateLimit {
 			return nil, ErrRateLimitExceeded
 		}
@@ -106,16 +352,26 @@ func (ft *FindATender) fetchURI(
 		if err != nil {
 			return nil, err
 		}
+		if res.StatusCode() == 429 {
+			closeBody(res)
+			return nil, ErrRateLimitExceeded
+		}
+	}
+
+	if res.StatusCode() != 200 {
+		closeBody(res)
+		return nil, fmt.Errorf("%w: %s", ErrUnexpectedStatus, res.Status())
 	}
 
 	return res, nil
 }
 
+// Perform a fetch against the API with some sanitised default options.
 func (ft *FindATender) fetchWithOpts(
 	opts RequestOptions,
 ) (*resty.Response, error) {
 	if opts.Since.IsZero() {
-		opts.Since = time.Now().AddDate(0, 0, 1)
+		opts.Since = time.Now().AddDate(0, 0, -1)
 	}
 	if opts.Limit == 0 {
 		opts.Limit = 100
@@ -131,8 +387,11 @@ func (ft *FindATender) fetchWithOpts(
 	return ft.fetchURI(uri, opts.AllowRateLimit)
 }
 
+// If the response provided a "Next" link, perform another API call to get the
+// next page. Error if there is no "Next" link - the caller can decide if they
+// have attempted to fetch next in error or not.
 func (ft *FindATender) fetchNext(
-	pkg ocds.ReleasePackage,
+	pkg *ocds.ReleasePackage,
 	allowRateLimit bool,
 ) (*resty.Response, error) {
 	if pkg.Links == nil || len(pkg.Links.Next) == 0 {
@@ -141,26 +400,27 @@ func (ft *FindATender) fetchNext(
 	return ft.fetchURI(pkg.Links.Next, allowRateLimit)
 }
 
+// Make one call on the API. Do 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
 	}
-	defer res.Body.Close()
 
-	var relPkg ocds.ReleasePackage
-	decoder := json.NewDecoder(NewOCDSStreamCleaner(res.Body))
-	if err := decoder.Decode(&relPkg); err != nil {
+	pkg, err := safeUnpackReleases(res)
+	if err != nil {
 		return nil, err
 	}
 
-	if len(relPkg.Releases) == 0 {
+	if len(pkg.Releases) == 0 {
 		return nil, ErrNoReleases
 	}
 
-	return relPkg.Releases, nil
+	return pkg.Releases, nil
 }
 
+// Make a call on the API. If there are any "Next" links, call them in
+// succession until the end has been reached.
 func (ft *FindATender) fetchAll(opts RequestOptions) ([]ocds.Release, error) {
 	var releases []ocds.Release
 
@@ -168,30 +428,40 @@ func (ft *FindATender) fetchAll(opts RequestOptions) ([]ocds.Release, error) {
 	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 {
+	pkg, err := safeUnpackReleases(res)
+	if err != nil {
 		return nil, err
 	}
-	releases = append(releases, relPkg.Releases...)
+	releases = append(releases, pkg.Releases...)
 
-	for true {
-		res, err := ft.fetchNext(relPkg, opts.AllowRateLimit)
+	// Next links already followed, so a page that links back to an earlier
+	// one cannot loop forever.
+	seen := make(map[string]bool)
+	for {
+		if pkg.Links != nil && len(pkg.Links.Next) > 0 {
+			if seen[pkg.Links.Next] {
+				slog.Warn(fmt.Sprintf(
+					"Stopping at repeated next link: %s", pkg.Links.Next,
+				))
+				break
+			}
+			seen[pkg.Links.Next] = true
+		}
+
+		res, err := ft.fetchNext(pkg, 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 {
+		pkg, err = safeUnpackReleases(res)
+		if err != nil {
 			return nil, err
 		}
-		releases = append(releases, relPkg.Releases...)
+		releases = append(releases, pkg.Releases...)
 	}
 
 	if len(releases) == 0 {