Implement warmup and fetch retry in the weatherapi adapter

2026-07-02 11:33:16 -05:00
parent dc11e08e22
commit 27506168f8
4 changed files with 409 additions and 30 deletions
--- a/internal/adapters/weatherapi/client.go
+++ b/internal/adapters/weatherapi/client.go
@@ -24,6 +24,12 @@ import (
 const (
 	convectiveOutlooksEndpoint  = "/outlooks/convective"
 	sourceSPCConvectiveOutlooks = "spc_convective_outlooks"
+
+	defaultWarmupEndpoint  = "/conditions/current"
+	defaultWarmupAttempts  = 3
+	defaultWarmupDelay     = time.Second
+	defaultFetchAttempts   = 2
+	defaultFetchRetryDelay = time.Second
 )

 type Client struct {
@@ -35,6 +41,12 @@ type Client struct {
 	precision     int
 	missingSource config.MissingSourceConfig
 	now           func() time.Time
+
+	warmupEndpoint  string
+	warmupAttempts  int
+	warmupDelay     time.Duration
+	fetchAttempts   int
+	fetchRetryDelay time.Duration
 }

 type Option func(*Client)
@@ -80,7 +92,12 @@ func New(cfg config.Config, opts ...Option) (*Client, error) {
 			Default: cfg.MissingSource.Default,
 			Sources: cfg.MissingSource.Sources,
 		},
-		now: time.Now,
+		now:             time.Now,
+		warmupEndpoint:  defaultWarmupEndpoint,
+		warmupAttempts:  defaultWarmupAttempts,
+		warmupDelay:     defaultWarmupDelay,
+		fetchAttempts:   defaultFetchAttempts,
+		fetchRetryDelay: defaultFetchRetryDelay,
 	}
 	for _, opt := range opts {
 		opt(client)
@@ -89,6 +106,10 @@ func New(cfg config.Config, opts ...Option) (*Client, error) {
 }

 func (c *Client) FetchBundle(ctx context.Context) (*weatherdata.Bundle, error) {
+	if err := c.warmup(ctx); err != nil {
+		return nil, err
+	}
+
 	fetchedAt := c.now()
 	builder := bundleBuilder{
 		client:    c,
@@ -397,24 +418,9 @@ type envelope struct {
 }

 func (c *Client) fetch(ctx context.Context, sourceName string, endpoint string, opts queryOptions) (json.RawMessage, weatherdata.Source, error) {
-	reqURL := c.endpointURL(endpoint, opts)
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil)
+	reqURL, body, err := c.fetchHTTP(ctx, endpoint, opts)
 	if err != nil {
-		return nil, weatherdata.Source{}, fmt.Errorf("create request for %s: %w", endpoint, err)
-	}
-
-	resp, err := c.httpClient.Do(req)
-	if err != nil {
-		return nil, weatherdata.Source{}, fmt.Errorf("fetch %s: %w", endpoint, err)
-	}
-	defer resp.Body.Close()
-
-	body, err := io.ReadAll(io.LimitReader(resp.Body, 10<<20))
-	if err != nil {
-		return nil, weatherdata.Source{}, fmt.Errorf("read %s response: %w", endpoint, err)
-	}
-	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		return nil, weatherdata.Source{}, fmt.Errorf("fetch %s: unexpected HTTP status %d: %s", endpoint, resp.StatusCode, strings.TrimSpace(string(body)))
+		return nil, weatherdata.Source{}, err
 	}

 	var env envelope
@@ -440,6 +446,169 @@ func (c *Client) fetch(ctx context.Context, sourceName string, endpoint string,
 	return env.Data, source, nil
 }

+func (c *Client) warmup(ctx context.Context) error {
+	endpoint := c.warmupEndpoint
+	if strings.TrimSpace(endpoint) == "" {
+		endpoint = defaultWarmupEndpoint
+	}
+	attempts := positiveAttemptCount(c.warmupAttempts)
+	var lastErr error
+	for attempt := 1; attempt <= attempts; attempt++ {
+		if err := ctx.Err(); err != nil {
+			return fmt.Errorf("warm up weather API via %s: %w", endpoint, err)
+		}
+		if err := c.warmupOnce(ctx, endpoint); err != nil {
+			lastErr = err
+		} else {
+			return nil
+		}
+		if attempt == attempts {
+			break
+		}
+		if err := waitForRetry(ctx, c.warmupDelay); err != nil {
+			return fmt.Errorf("warm up weather API via %s after %d attempt(s): %w", endpoint, attempt, err)
+		}
+	}
+	return fmt.Errorf("warm up weather API via %s failed after %d attempts: %w", endpoint, attempts, lastErr)
+}
+
+func (c *Client) warmupOnce(ctx context.Context, endpoint string) error {
+	reqURL := c.endpointURL(endpoint, queryOptions{precision: true})
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil)
+	if err != nil {
+		return fmt.Errorf("create request for %s: %w", endpoint, err)
+	}
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("fetch %s: %w", endpoint, err)
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 10<<20))
+	if err != nil {
+		return fmt.Errorf("read %s response: %w", endpoint, err)
+	}
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return fmt.Errorf("fetch %s: unexpected HTTP status %d: %s", endpoint, resp.StatusCode, strings.TrimSpace(string(body)))
+	}
+	return nil
+}
+
+func (c *Client) fetchHTTP(ctx context.Context, endpoint string, opts queryOptions) (*url.URL, []byte, error) {
+	attempts := positiveAttemptCount(c.fetchAttempts)
+	var lastErr error
+	var lastRetryable bool
+	for attempt := 1; attempt <= attempts; attempt++ {
+		if err := ctx.Err(); err != nil {
+			return nil, nil, fmt.Errorf("fetch %s: %w", endpoint, err)
+		}
+		reqURL, body, err := c.fetchHTTPOnce(ctx, endpoint, opts)
+		if err == nil {
+			return reqURL, body, nil
+		}
+		lastErr = err
+		lastRetryable = isRetryableRequestError(err)
+		if !lastRetryable || attempt == attempts {
+			break
+		}
+		if err := waitForRetry(ctx, c.fetchRetryDelay); err != nil {
+			return nil, nil, fmt.Errorf("fetch %s retry delay after attempt %d: %w", endpoint, attempt, err)
+		}
+	}
+	if lastRetryable {
+		return nil, nil, fmt.Errorf("fetch %s failed after %d attempts: %w", endpoint, attempts, lastErr)
+	}
+	return nil, nil, lastErr
+}
+
+func (c *Client) fetchHTTPOnce(ctx context.Context, endpoint string, opts queryOptions) (*url.URL, []byte, error) {
+	reqURL := c.endpointURL(endpoint, opts)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil)
+	if err != nil {
+		return nil, nil, fmt.Errorf("create request for %s: %w", endpoint, err)
+	}
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		err = fmt.Errorf("fetch %s: %w", endpoint, err)
+		if ctx.Err() != nil {
+			return reqURL, nil, err
+		}
+		return reqURL, nil, retryableRequestError{err: err}
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 10<<20))
+	if err != nil {
+		err = fmt.Errorf("read %s response: %w", endpoint, err)
+		if ctx.Err() != nil {
+			return reqURL, nil, err
+		}
+		return reqURL, nil, retryableRequestError{err: err}
+	}
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		err := fmt.Errorf("fetch %s: unexpected HTTP status %d: %s", endpoint, resp.StatusCode, strings.TrimSpace(string(body)))
+		if isRetryableHTTPStatus(resp.StatusCode) {
+			return reqURL, nil, retryableRequestError{err: err}
+		}
+		return reqURL, nil, err
+	}
+	return reqURL, body, nil
+}
+
+type retryableRequestError struct {
+	err error
+}
+
+func (e retryableRequestError) Error() string {
+	return e.err.Error()
+}
+
+func (e retryableRequestError) Unwrap() error {
+	return e.err
+}
+
+func isRetryableRequestError(err error) bool {
+	_, ok := err.(retryableRequestError)
+	return ok
+}
+
+func isRetryableHTTPStatus(status int) bool {
+	switch status {
+	case http.StatusRequestTimeout,
+		http.StatusTooManyRequests,
+		http.StatusInternalServerError,
+		http.StatusBadGateway,
+		http.StatusServiceUnavailable,
+		http.StatusGatewayTimeout:
+		return true
+	default:
+		return false
+	}
+}
+
+func waitForRetry(ctx context.Context, delay time.Duration) error {
+	if delay <= 0 {
+		return ctx.Err()
+	}
+	timer := time.NewTimer(delay)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
+func positiveAttemptCount(attempts int) int {
+	if attempts < 1 {
+		return 1
+	}
+	return attempts
+}
+
 func isJSONNull(raw json.RawMessage) bool {
 	return bytes.Equal(bytes.TrimSpace(raw), []byte("null"))
 }