weatherreporter/internal/adapters/weatherapi/client.go

// Package weatherapi adapts the internal weather API to weather data bundles.
package weatherapi

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"path"
	"strconv"
	"strings"
	"time"

	"gitea.maximumdirect.net/eric/weatherreporter/internal/config"
	"gitea.maximumdirect.net/eric/weatherreporter/internal/fileutil"
	"gitea.maximumdirect.net/eric/weatherreporter/internal/weatherdata"
)

const (
	convectiveOutlooksEndpoint  = "/outlooks/convective"
	sourceSPCConvectiveOutlooks = config.MissingSourceSPCConvectiveOutlooks

	defaultWarmupEndpoint  = "/conditions/current"
	defaultWarmupAttempts  = 3
	defaultWarmupDelay     = time.Second
	defaultFetchAttempts   = 2
	defaultFetchRetryDelay = time.Second
	maxResponseBodyBytes   = 10 << 20
)

var errResponseBodyTooLarge = errors.New("response exceeds 10 MiB limit")

type Client struct {
	baseURL       *url.URL
	httpClient    *http.Client
	units         string
	format        string
	timezone      string
	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)

func WithHTTPClient(httpClient *http.Client) Option {
	return func(c *Client) {
		if httpClient != nil {
			c.httpClient = httpClient
		}
	}
}

func WithClock(now func() time.Time) Option {
	return func(c *Client) {
		if now != nil {
			c.now = now
		}
	}
}

func New(cfg config.Config, opts ...Option) (*Client, error) {
	if strings.TrimSpace(cfg.WeatherAPI.BaseURL) == "" {
		return nil, fmt.Errorf("weather_api.base_url is required")
	}
	baseURL, err := url.Parse(cfg.WeatherAPI.BaseURL)
	if err != nil || baseURL.Scheme == "" || baseURL.Host == "" {
		return nil, fmt.Errorf("weather_api.base_url must be an absolute URL")
	}
	if !strings.EqualFold(baseURL.Scheme, "http") && !strings.EqualFold(baseURL.Scheme, "https") {
		return nil, fmt.Errorf("weather_api.base_url must use http or https")
	}

	timeout := cfg.WeatherAPI.Timeout
	if timeout <= 0 {
		timeout = 10 * time.Second
	}

	client := &Client{
		baseURL:    baseURL,
		httpClient: &http.Client{Timeout: timeout},
		units:      cfg.WeatherAPI.Units,
		format:     cfg.WeatherAPI.Format,
		timezone:   cfg.WeatherAPI.Timezone,
		precision:  cfg.WeatherAPI.Precision,
		missingSource: config.MissingSourceConfig{
			Default: cfg.MissingSource.Default,
			Sources: cfg.MissingSource.Sources,
		},
		now:             time.Now,
		warmupEndpoint:  defaultWarmupEndpoint,
		warmupAttempts:  defaultWarmupAttempts,
		warmupDelay:     defaultWarmupDelay,
		fetchAttempts:   defaultFetchAttempts,
		fetchRetryDelay: defaultFetchRetryDelay,
	}
	for _, opt := range opts {
		opt(client)
	}
	return client, nil
}

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,
		bundle:    &weatherdata.Bundle{FetchedAt: fetchedAt},
		fetchedAt: fetchedAt,
	}

	if err := builder.fetchObservation(ctx); err != nil {
		return nil, err
	}
	if err := builder.fetchCurrent(ctx); err != nil {
		return nil, err
	}
	if err := builder.fetchHourly(ctx); err != nil {
		return nil, err
	}
	if err := builder.fetchNarrative(ctx); err != nil {
		return nil, err
	}
	if err := builder.fetchAlerts(ctx); err != nil {
		return nil, err
	}
	if err := builder.fetchDiscussion(ctx); err != nil {
		return nil, err
	}
	if err := builder.fetchWeatherStory(ctx); err != nil {
		return nil, err
	}
	if err := builder.fetchSPCConvectiveOutlooks(ctx); err != nil {
		return nil, err
	}

	return builder.bundle, nil
}

type bundleBuilder struct {
	client    *Client
	bundle    *weatherdata.Bundle
	fetchedAt time.Time
}

type sourceRequest struct {
	name           string
	endpoint       string
	query          queryOptions
	missingMessage string
	required       bool
	decodeLabel    string
}

type fetchedSource struct {
	raw    json.RawMessage
	source weatherdata.Source
}

func (b *bundleBuilder) fetchObservation(ctx context.Context) error {
	var observation weatherdata.Observation
	fetched, ok, err := b.fetchDecodedSource(ctx, sourceRequest{
		name:           config.MissingSourceObservations,
		endpoint:       "/observations",
		query:          queryOptions{precision: true},
		missingMessage: "observation data is missing",
	}, &observation)
	if err != nil || !ok {
		return err
	}
	source := fetched.source
	source.IssuedAt = &observation.Timestamp
	b.bundle.Observation = &observation
	b.addSource(source)
	return nil
}

func (b *bundleBuilder) fetchCurrent(ctx context.Context) error {
	var current weatherdata.Current
	fetched, ok, err := b.fetchDecodedSource(ctx, sourceRequest{
		name:           config.MissingSourceCurrent,
		endpoint:       "/conditions/current",
		query:          queryOptions{precision: true},
		missingMessage: "current conditions data is missing",
	}, &current)
	if err != nil || !ok {
		return err
	}
	source := fetched.source
	b.bundle.Current = &current
	b.addSource(source)
	return nil
}

func (b *bundleBuilder) fetchHourly(ctx context.Context) error {
	var hourly weatherdata.ForecastRun
	fetched, ok, err := b.fetchDecodedSource(ctx, sourceRequest{
		name:           "hourly",
		endpoint:       "/forecast/hourly",
		query:          queryOptions{precision: true, timezone: true},
		missingMessage: "hourly forecast data is missing",
		required:       true,
		decodeLabel:    "hourly forecast",
	}, &hourly)
	if err != nil || !ok {
		return err
	}
	source := fetched.source
	if len(hourly.Periods) == 0 {
		return fmt.Errorf("hourly forecast from %s contains no periods", source.Endpoint)
	}
	for i, period := range hourly.Periods {
		if !period.HasUsableTimeBounds() {
			return fmt.Errorf("hourly forecast from %s has unusable time bounds for period %d", source.Endpoint, i+1)
		}
		if !period.HasValidPrecipitationProbability() {
			return fmt.Errorf("hourly forecast from %s has invalid precipitation probability for period %d", source.Endpoint, i+1)
		}
	}
	source.IssuedAt = &hourly.IssuedAt
	source.UpdatedAt = hourly.UpdatedAt
	b.bundle.Hourly = &hourly
	b.addSource(source)
	return nil
}

func (b *bundleBuilder) fetchNarrative(ctx context.Context) error {
	var narrative weatherdata.ForecastRun
	fetched, ok, err := b.fetchDecodedSource(ctx, sourceRequest{
		name:           config.MissingSourceNarrative,
		endpoint:       "/forecast/narrative",
		query:          queryOptions{precision: true, timezone: true},
		missingMessage: "narrative forecast data is missing",
	}, &narrative)
	if err != nil || !ok {
		return err
	}
	source := fetched.source
	source.IssuedAt = &narrative.IssuedAt
	source.UpdatedAt = narrative.UpdatedAt
	b.bundle.Narrative = &narrative
	b.addSource(source)
	return nil
}

func (b *bundleBuilder) fetchAlerts(ctx context.Context) error {
	raw, source, err := b.client.fetch(ctx, config.MissingSourceAlerts, "/alerts/active", queryOptions{allowNull: true})
	if err != nil {
		return err
	}
	if raw == nil {
		return b.handleMissing(&source, "active alerts data is missing", false)
	}
	if isJSONNull(raw) {
		b.bundle.Alerts = &weatherdata.AlertRun{Raw: append(json.RawMessage(nil), raw...)}
		b.addSource(source)
		return nil
	}
	var alerts weatherdata.AlertRun
	if err := decodeSource(raw, &alerts); err != nil {
		return b.handleMalformed(&source, err, sourceRequest{name: config.MissingSourceAlerts})
	}
	alerts.Raw = append(json.RawMessage(nil), raw...)
	if alerts.AsOf != nil {
		source.IssuedAt = alerts.AsOf
	}
	b.bundle.Alerts = &alerts
	b.addSource(source)
	return nil
}

func (b *bundleBuilder) fetchDiscussion(ctx context.Context) error {
	var discussion weatherdata.Discussion
	fetched, ok, err := b.fetchDecodedSource(ctx, sourceRequest{
		name:           config.MissingSourceDiscussion,
		endpoint:       "/discussion",
		query:          queryOptions{timezone: true},
		missingMessage: "forecast discussion data is missing",
	}, &discussion)
	if err != nil || !ok {
		return err
	}
	source := fetched.source
	source.IssuedAt = &discussion.IssuedAt
	source.UpdatedAt = discussion.UpdatedAt
	b.bundle.Discussion = &discussion
	b.addSource(source)
	return nil
}

func (b *bundleBuilder) fetchWeatherStory(ctx context.Context) error {
	var story weatherdata.WeatherStory
	request := sourceRequest{
		name:           config.MissingSourceWeatherStory,
		endpoint:       "/weatherstories/latest",
		query:          queryOptions{omitUnits: true},
		missingMessage: "NWS weather story data is missing",
	}
	fetched, ok, err := b.fetchDecodedSource(ctx, request, &story)
	if err != nil || !ok {
		return err
	}
	source := fetched.source
	if !story.HasUsableContent() {
		return b.handleMalformed(&source, fmt.Errorf("weather story has no usable content"), request)
	}
	if !story.StartTime.IsZero() {
		source.IssuedAt = &story.StartTime
	}
	source.UpdatedAt = story.UpdatedAt
	b.bundle.WeatherStory = &story
	b.addSource(source)
	return nil
}

func (b *bundleBuilder) fetchSPCConvectiveOutlooks(ctx context.Context) error {
	var run weatherdata.ConvectiveOutlookRun
	fetched, ok, err := b.fetchDecodedSource(ctx, sourceRequest{
		name:           sourceSPCConvectiveOutlooks,
		endpoint:       convectiveOutlooksEndpoint,
		query:          queryOptions{timezone: true, omitUnits: true},
		missingMessage: "SPC convective outlook data is missing",
	}, &run)
	if err != nil || !ok {
		return err
	}
	source := fetched.source
	if run.IssuedAt != nil {
		source.IssuedAt = run.IssuedAt
	} else {
		source.IssuedAt = run.AsOf
	}
	source.UpdatedAt = run.UpdatedAt
	b.bundle.SPCConvectiveOutlooks = &run
	b.addSource(source)
	return nil
}

func (b *bundleBuilder) fetchDecodedSource(ctx context.Context, request sourceRequest, target any) (fetchedSource, bool, error) {
	fetched, ok, err := b.fetchSource(ctx, request)
	if err != nil || !ok {
		return fetchedSource{}, false, err
	}
	if err := decodeSource(fetched.raw, target); err != nil {
		return fetchedSource{}, false, b.handleMalformed(&fetched.source, err, request)
	}
	return fetched, true, nil
}

func (b *bundleBuilder) fetchSource(ctx context.Context, request sourceRequest) (fetchedSource, bool, error) {
	raw, source, err := b.client.fetch(ctx, request.name, request.endpoint, request.query)
	if err != nil {
		return fetchedSource{}, false, err
	}
	if raw == nil {
		return fetchedSource{}, false, b.handleMissing(&source, request.missingMessage, request.required)
	}
	return fetchedSource{raw: raw, source: source}, true, nil
}

func (b *bundleBuilder) handleMissing(source *weatherdata.Source, message string, required bool) error {
	source.Missing = true
	if required {
		return fmt.Errorf("%s from %s is required", message, source.Endpoint)
	}
	return b.applyMissingPolicy(source, "missing_source", message)
}

func (b *bundleBuilder) handleMalformed(source *weatherdata.Source, err error, request sourceRequest) error {
	if request.required {
		label := request.name
		if request.decodeLabel != "" {
			label = request.decodeLabel
		}
		return fmt.Errorf("decode %s from %s: %w", label, source.Endpoint, err)
	}
	source.Missing = true
	return b.applyMissingPolicy(source, "malformed_source", fmt.Sprintf("malformed %s data: %v", source.Name, err))
}

func (b *bundleBuilder) applyMissingPolicy(source *weatherdata.Source, code string, message string) error {
	policy := b.client.policyFor(source.Name)
	if policy == config.MissingSourceError {
		return fmt.Errorf("%s: %s", source.Name, message)
	}
	if policy == config.MissingSourceWarn {
		warning := weatherdata.SourceWarning{
			Source:             source.Name,
			Code:               code,
			Severity:           "warning",
			Message:            message,
			Endpoint:           source.Endpoint,
			CompletenessImpact: "source omitted from bundle",
		}
		source.Warnings = append(source.Warnings, warning)
		b.bundle.Warnings = append(b.bundle.Warnings, warning)
	}
	b.addSource(*source)
	return nil
}

func (b *bundleBuilder) addSource(source weatherdata.Source) {
	b.bundle.Sources = append(b.bundle.Sources, source)
}

func (c *Client) policyFor(source string) config.MissingSourcePolicy {
	if policy, ok := c.missingSource.Sources[source]; ok {
		return policy
	}
	return c.missingSource.Default
}

type queryOptions struct {
	precision bool
	timezone  bool
	allowNull bool
	omitUnits bool
}

type envelope struct {
	Data json.RawMessage `json:"data"`
}

func (c *Client) fetch(ctx context.Context, sourceName string, endpoint string, opts queryOptions) (json.RawMessage, weatherdata.Source, error) {
	reqURL, body, err := c.fetchHTTP(ctx, endpoint, opts)
	if err != nil {
		return nil, weatherdata.Source{}, err
	}

	var env envelope
	if err := json.Unmarshal(body, &env); err != nil {
		return nil, weatherdata.Source{}, fmt.Errorf("decode %s envelope: %w", endpoint, err)
	}

	source := weatherdata.Source{
		Name:      sourceName,
		Endpoint:  endpoint,
		Query:     queryMap(reqURL.Query()),
		FetchedAt: c.now(),
	}
	if len(env.Data) == 0 || (isJSONNull(env.Data) && !opts.allowNull) {
		source.Missing = true
		return nil, source, nil
	}
	hash, err := sourceHash(env.Data)
	if err != nil {
		return env.Data, source, nil
	}
	source.DataSHA256 = hash
	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
	var lastRetryable bool
	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
			lastRetryable = isRetryableRequestError(err)
		} else {
			return nil
		}
		if !lastRetryable || 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)
		}
	}
	if !lastRetryable {
		return lastErr
	}
	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 {
		err = fmt.Errorf("fetch %s: %w", endpoint, err)
		if ctx.Err() != nil {
			return err
		}
		return retryableRequestError{err: err}
	}
	defer resp.Body.Close()

	_, err = readResponseBody(resp.Body)
	if err != nil {
		return responseReadError(ctx, endpoint, err)
	}
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		err := fmt.Errorf("fetch %s: unexpected HTTP status %d", endpoint, resp.StatusCode)
		if isRetryableHTTPStatus(resp.StatusCode) {
			return retryableRequestError{err: err}
		}
		return err
	}
	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 := readResponseBody(resp.Body)
	if err != nil {
		return reqURL, nil, responseReadError(ctx, endpoint, err)
	}
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		err := fmt.Errorf("fetch %s: unexpected HTTP status %d", endpoint, resp.StatusCode)
		if isRetryableHTTPStatus(resp.StatusCode) {
			return reqURL, nil, retryableRequestError{err: err}
		}
		return reqURL, nil, err
	}
	return reqURL, body, nil
}

func readResponseBody(body io.Reader) ([]byte, error) {
	data, err := io.ReadAll(io.LimitReader(body, maxResponseBodyBytes+1))
	if err != nil {
		return nil, err
	}
	if int64(len(data)) > maxResponseBodyBytes {
		return nil, errResponseBodyTooLarge
	}
	return data, nil
}

func responseReadError(ctx context.Context, endpoint string, err error) error {
	err = fmt.Errorf("read %s response: %w", endpoint, err)
	if errors.Is(err, errResponseBodyTooLarge) || ctx.Err() != nil {
		return err
	}
	return retryableRequestError{err: err}
}

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"))
}

func (c *Client) endpointURL(endpoint string, opts queryOptions) *url.URL {
	reqURL := *c.baseURL
	reqURL.Path = path.Join(c.baseURL.Path, endpoint)
	query := reqURL.Query()
	query.Set("format", c.format)
	if !opts.omitUnits {
		query.Set("units", c.units)
	}
	if opts.precision {
		query.Set("precision", strconv.Itoa(c.precision))
	}
	if opts.timezone {
		query.Set("tz", c.timezone)
	}
	reqURL.RawQuery = query.Encode()
	return &reqURL
}

func queryMap(values url.Values) map[string]string {
	if len(values) == 0 {
		return nil
	}
	out := make(map[string]string, len(values))
	for key, value := range values {
		if len(value) > 0 {
			out[key] = value[0]
		}
	}
	return out
}

func decodeSource(raw json.RawMessage, target any) error {
	if err := json.Unmarshal(raw, target); err != nil {
		return err
	}
	return nil
}

func sourceHash(raw json.RawMessage) (string, error) {
	var compact bytes.Buffer
	if err := json.Compact(&compact, raw); err != nil {
		return "", err
	}
	sum := sha256.Sum256(compact.Bytes())
	return hex.EncodeToString(sum[:]), nil
}

func SaveBundle(path string, bundle *weatherdata.Bundle) error {
	if err := fileutil.WriteJSONAtomic(path, bundle); err != nil {
		return fmt.Errorf("save bundle: %w", err)
	}
	return nil
}