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" "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") var ErrUnexpectedStatus = errors.New("unexpected response status") type RequestOptions struct { Since time.Time Limit int 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() select { case <-timer.C: return nil case <-ctx.Done(): return ctx.Err() } } // 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 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 } // 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, ) (*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 { closeBody(res) 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 } 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) } 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) } // 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, allowRateLimit bool, ) (*resty.Response, error) { if pkg.Links == nil || len(pkg.Links.Next) == 0 { return nil, ErrNoNext } 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 } pkg, err := safeUnpackReleases(res) if err != nil { return nil, err } if len(pkg.Releases) == 0 { return nil, ErrNoReleases } 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 res, err := ft.fetchWithOpts(opts) if err != nil { return nil, err } pkg, err := safeUnpackReleases(res) if err != nil { return nil, err } releases = append(releases, pkg.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 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 } pkg, err = safeUnpackReleases(res) if err != nil { return nil, err } releases = append(releases, pkg.Releases...) } if len(releases) == 0 { return nil, ErrNoReleases } return releases, nil }