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"
	"sync"
	"time"

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

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

	defaultWarmupEndpoint  = currentConditionsEndpoint
	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) {
	warmup, err := c.warmup(ctx)
	if err != nil {
		return nil, err
	}

	fetchedAt := c.now()
	builder := bundleBuilder{
		client: c,
		bundle: &weatherdata.Bundle{FetchedAt: fetchedAt},
	}

	for _, acquired := range builder.acquireSources(ctx, warmup) {
		if err := ctx.Err(); err != nil {
			return nil, fmt.Errorf("fetch weather API sources: %w", err)
		}
		if err := builder.mergeSource(acquired); err != nil {
			return nil, err
		}
	}

	return builder.bundle, nil
}

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

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

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

type warmupResponse struct {
	endpoint   string
	requestURL *url.URL
	body       []byte
	fetchedAt  time.Time
}

type sourceAcquisition struct {
	request    sourceRequest
	fetched    fetchedSource
	err        error
	warmup     warmupResponse
	usesWarmup bool
}

func (b *bundleBuilder) acquireSources(ctx context.Context, warmup warmupResponse) []sourceAcquisition {
	sources := []sourceAcquisition{
		{request: sourceRequest{name: config.MissingSourceObservations, endpoint: "/observations", query: queryOptions{precision: true}, missingMessage: "observation data is missing"}},
		{request: currentConditionsRequest()},
		{request: sourceRequest{name: "hourly", endpoint: "/forecast/hourly", query: queryOptions{precision: true, timezone: true}, missingMessage: "hourly forecast data is missing", required: true, decodeLabel: "hourly forecast"}},
		{request: sourceRequest{name: config.MissingSourceNarrative, endpoint: "/forecast/narrative", query: queryOptions{precision: true, timezone: true}, missingMessage: "narrative forecast data is missing"}},
		{request: sourceRequest{name: config.MissingSourceAlerts, endpoint: "/alerts/active", query: queryOptions{allowNull: true}, missingMessage: "active alerts data is missing"}},
		{request: sourceRequest{name: config.MissingSourceDiscussion, endpoint: "/discussion", query: queryOptions{timezone: true}, missingMessage: "forecast discussion data is missing"}},
		{request: sourceRequest{name: config.MissingSourceWeatherStory, endpoint: "/weatherstories/latest", query: queryOptions{omitUnits: true}, missingMessage: "NWS weather story data is missing"}},
		{request: sourceRequest{name: sourceSPCConvectiveOutlooks, endpoint: convectiveOutlooksEndpoint, query: queryOptions{timezone: true, omitUnits: true}, missingMessage: "SPC convective outlook data is missing"}},
	}
	if warmup.endpoint == currentConditionsEndpoint {
		sources[1].warmup = warmup
		sources[1].usesWarmup = true
	}

	var group sync.WaitGroup
	for i := range sources {
		if sources[i].usesWarmup {
			continue
		}
		group.Add(1)
		go func(index int) {
			defer group.Done()
			request := sources[index].request
			raw, source, err := b.client.fetch(ctx, request.name, request.endpoint, request.query)
			sources[index].fetched = fetchedSource{raw: raw, source: source}
			sources[index].err = err
		}(i)
	}
	group.Wait()
	return sources
}

func (b *bundleBuilder) mergeSource(acquired sourceAcquisition) error {
	switch acquired.request.name {
	case config.MissingSourceObservations:
		return b.fetchObservation(acquired)
	case config.MissingSourceCurrent:
		return b.fetchCurrent(acquired)
	case "hourly":
		return b.fetchHourly(acquired)
	case config.MissingSourceNarrative:
		return b.fetchNarrative(acquired)
	case config.MissingSourceAlerts:
		return b.fetchAlerts(acquired)
	case config.MissingSourceDiscussion:
		return b.fetchDiscussion(acquired)
	case config.MissingSourceWeatherStory:
		return b.fetchWeatherStory(acquired)
	case sourceSPCConvectiveOutlooks:
		return b.fetchSPCConvectiveOutlooks(acquired)
	default:
		return fmt.Errorf("merge unknown weather source %q", acquired.request.name)
	}
}

func (b *bundleBuilder) fetchObservation(acquired sourceAcquisition) error {
	var observation weatherdata.Observation
	fetched, ok, err := b.fetchDecodedSource(acquired, &observation)
	if err != nil || !ok {
		return err
	}
	source := fetched.source
	source.IssuedAt = &observation.Timestamp
	b.bundle.Observation = &observation
	b.addSource(source)
	return nil
}

func currentConditionsRequest() sourceRequest {
	return sourceRequest{
		name:           config.MissingSourceCurrent,
		endpoint:       currentConditionsEndpoint,
		query:          queryOptions{precision: true},
		missingMessage: "current conditions data is missing",
	}
}

func (b *bundleBuilder) fetchCurrent(acquired sourceAcquisition) error {
	var current weatherdata.Current
	fetched, ok, err := b.fetchDecodedSource(acquired, &current)
	if err != nil || !ok {
		return err
	}
	source := fetched.source
	b.bundle.Current = &current
	b.addSource(source)
	return nil
}

func (b *bundleBuilder) fetchHourly(acquired sourceAcquisition) error {
	var hourly weatherdata.ForecastRun
	fetched, ok, err := b.fetchDecodedSource(acquired, &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(acquired sourceAcquisition) error {
	var narrative weatherdata.ForecastRun
	fetched, ok, err := b.fetchDecodedSource(acquired, &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(acquired sourceAcquisition) error {
	fetched, ok, err := b.fetchSource(acquired)
	if err != nil || !ok {
		return err
	}
	raw, source := fetched.raw, fetched.source
	if isJSONNull(raw) {
		b.bundle.Alerts = &weatherdata.AlertRun{}
		b.addSource(source)
		return nil
	}
	var alerts weatherdata.AlertRun
	if err := decodeSource(raw, &alerts); err != nil {
		return b.handleMalformed(&source, err, acquired.request)
	}
	if alerts.AsOf != nil {
		source.IssuedAt = alerts.AsOf
	}
	b.bundle.Alerts = &alerts
	b.addSource(source)
	return nil
}

func (b *bundleBuilder) fetchDiscussion(acquired sourceAcquisition) error {
	var discussion weatherdata.Discussion
	fetched, ok, err := b.fetchDecodedSource(acquired, &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(acquired sourceAcquisition) error {
	var story weatherdata.WeatherStory
	fetched, ok, err := b.fetchDecodedSource(acquired, &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"), acquired.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(acquired sourceAcquisition) error {
	var run weatherdata.ConvectiveOutlookRun
	fetched, ok, err := b.fetchDecodedSource(acquired, &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(acquired sourceAcquisition, target any) (fetchedSource, bool, error) {
	fetched, ok, err := b.fetchSource(acquired)
	if err != nil || !ok {
		return fetchedSource{}, false, err
	}
	return b.decodeFetchedSource(fetched, acquired.request, target)
}

func (b *bundleBuilder) decodeFetchedSource(fetched fetchedSource, request sourceRequest, target any) (fetchedSource, bool, error) {
	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(acquired sourceAcquisition) (fetchedSource, bool, error) {
	if acquired.usesWarmup {
		raw, source, err := b.client.decodeSourceResponse(acquired.request.name, acquired.request.endpoint, acquired.request.query, acquired.warmup.requestURL, acquired.warmup.body, acquired.warmup.fetchedAt)
		if err != nil {
			return fetchedSource{}, false, err
		}
		acquired.fetched = fetchedSource{raw: raw, source: source}
	} else if acquired.err != nil {
		return fetchedSource{}, false, acquired.err
	}
	if acquired.fetched.raw == nil {
		return fetchedSource{}, false, b.handleMissing(&acquired.fetched.source, acquired.request.missingMessage, acquired.request.required)
	}
	return acquired.fetched, 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
	}
	return c.decodeSourceResponse(sourceName, endpoint, opts, reqURL, body, c.now())
}

func (c *Client) decodeSourceResponse(sourceName string, endpoint string, opts queryOptions, reqURL *url.URL, body []byte, fetchedAt time.Time) (json.RawMessage, weatherdata.Source, error) {
	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: fetchedAt,
	}
	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) (warmupResponse, 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 warmupResponse{}, fmt.Errorf("warm up weather API via %s: %w", endpoint, err)
		}
		reqURL, body, err := c.warmupOnce(ctx, endpoint)
		if err != nil {
			lastErr = err
			lastRetryable = isRetryableRequestError(err)
		} else {
			return warmupResponse{endpoint: endpoint, requestURL: reqURL, body: body, fetchedAt: c.now()}, nil
		}
		if !lastRetryable || attempt == attempts {
			break
		}
		if err := waitForRetry(ctx, c.warmupDelay); err != nil {
			return warmupResponse{}, fmt.Errorf("warm up weather API via %s after %d attempt(s): %w", endpoint, attempt, err)
		}
	}
	if !lastRetryable {
		return warmupResponse{}, lastErr
	}
	return warmupResponse{}, 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) (*url.URL, []byte, error) {
	return c.fetchHTTPOnce(ctx, endpoint, queryOptions{precision: true})
}

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
}