metadata-agregator/internal/provider/musicbrainz/client.go

package musicbrainz

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/rs/zerolog"
	"golang.org/x/time/rate"

	"github.com/metadata-agregator/internal/metrics"
)

const (
	baseURL     = "https://musicbrainz.org/ws/2"
	userAgent   = "MetadataAggregator/0.1.0 (https://github.com/metadata-agregator)"
	maxRetries  = 3
	baseBackoff = 2 * time.Second
)

type client struct {
	http    *http.Client
	limiter *rate.Limiter
}

func newClient() *client {
	return &client{
		http: &http.Client{
			Timeout: 30 * time.Second,
		},
		limiter: rate.NewLimiter(rate.Every(time.Second), 1),
	}
}

func (c *client) get(ctx context.Context, endpoint string, params url.Values) ([]byte, error) {
	var lastErr error

	for attempt := 0; attempt <= maxRetries; attempt++ {
		if err := c.waitForRateLimit(ctx); err != nil {
			return nil, err
		}

		req, err := c.buildRequest(ctx, endpoint, params)
		if err != nil {
			return nil, err
		}

		data, err := c.executeAndRead(ctx, req, endpoint)
		if err == nil {
			return data, nil
		}

		if err != ErrRateLimited {
			return nil, err
		}

		lastErr = err
		if attempt < maxRetries {
			backoff := baseBackoff * time.Duration(1<<attempt)
			zerolog.Ctx(ctx).Info().
				Int("attempt", attempt+1).
				Int("max_retries", maxRetries).
				Dur("backoff", backoff).
				Msg("rate limited, waiting before retry")

			select {
			case <-ctx.Done():
				return nil, ctx.Err()
			case <-time.After(backoff):
			}
		}
	}

	return nil, lastErr
}

func (c *client) waitForRateLimit(ctx context.Context) error {
	log := zerolog.Ctx(ctx)
	log.Trace().Msg("waiting for rate limiter")

	if err := c.limiter.Wait(ctx); err != nil {
		log.Debug().Err(err).Msg("rate limiter interrupted")
		return fmt.Errorf("rate limiter: %w", err)
	}
	return nil
}

func (c *client) buildRequest(ctx context.Context, endpoint string, params url.Values) (*http.Request, error) {
	if params == nil {
		params = url.Values{}
	}
	params.Set("fmt", "json")

	reqURL := fmt.Sprintf("%s/%s?%s", baseURL, endpoint, params.Encode())

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
	if err != nil {
		return nil, fmt.Errorf("create request: %w", err)
	}

	req.Header.Set("User-Agent", userAgent)
	req.Header.Set("Accept", "application/json")
	return req, nil
}

func (c *client) executeAndRead(ctx context.Context, req *http.Request, endpoint string) ([]byte, error) {
	log := zerolog.Ctx(ctx)
	start := time.Now()

	log.Trace().Str("url", req.URL.String()).Msg("sending HTTP request")

	resp, err := c.http.Do(req)
	if err != nil {
		duration := time.Since(start)
		log.Debug().Err(err).Str("endpoint", endpoint).Dur("duration", duration).Msg("HTTP request failed")
		metrics.ProviderRequests.WithLabelValues("musicbrainz", endpoint, "error").Inc()
		return nil, fmt.Errorf("do request: %w", err)
	}
	defer resp.Body.Close()

	duration := time.Since(start)
	metrics.ProviderLatency.WithLabelValues("musicbrainz", endpoint).Observe(duration.Seconds())

	return c.readResponse(ctx, resp, endpoint, duration)
}

func (c *client) readResponse(ctx context.Context, resp *http.Response, endpoint string, duration time.Duration) ([]byte, error) {
	log := zerolog.Ctx(ctx)

	switch resp.StatusCode {
	case http.StatusOK:
		log.Trace().Str("endpoint", endpoint).Dur("duration", duration).Msg("HTTP request succeeded")
		metrics.ProviderRequests.WithLabelValues("musicbrainz", endpoint, "ok").Inc()
		return io.ReadAll(resp.Body)

	case http.StatusBadRequest:
		body, _ := io.ReadAll(resp.Body)
		log.Warn().Str("endpoint", endpoint).Str("body", string(body)).Dur("duration", duration).Msg("HTTP 400 bad request")
		metrics.ProviderRequests.WithLabelValues("musicbrainz", endpoint, "bad_request").Inc()
		return nil, fmt.Errorf("%w: %s", ErrBadRequest, extractErrorMessage(body))

	case http.StatusNotFound:
		log.Debug().Str("endpoint", endpoint).Dur("duration", duration).Msg("HTTP 404")
		metrics.ProviderRequests.WithLabelValues("musicbrainz", endpoint, "not_found").Inc()
		return nil, ErrNotFound

	case http.StatusServiceUnavailable:
		log.Warn().Str("endpoint", endpoint).Dur("duration", duration).Msg("HTTP 503 rate limited by provider")
		metrics.ProviderRequests.WithLabelValues("musicbrainz", endpoint, "rate_limited").Inc()
		return nil, ErrRateLimited

	default:
		body, _ := io.ReadAll(resp.Body)
		log.Warn().Str("endpoint", endpoint).Int("status", resp.StatusCode).Str("body", string(body)).Dur("duration", duration).Msg("unexpected HTTP status")
		metrics.ProviderRequests.WithLabelValues("musicbrainz", endpoint, fmt.Sprintf("%d", resp.StatusCode)).Inc()
		return nil, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(body))
	}
}

func (c *client) lookup(ctx context.Context, entity, id string, inc []string) ([]byte, error) {
	zerolog.Ctx(ctx).Debug().Str("entity", entity).Str("id", id).Strs("includes", inc).Msg("provider lookup")

	params := url.Values{}
	if joined := joinIncludes(inc); joined != "" {
		params.Set("inc", joined)
	}

	return c.get(ctx, fmt.Sprintf("%s/%s", entity, id), params)
}

func (c *client) browse(ctx context.Context, entity, linkedEntity, linkedID string, limit, offset int, inc []string) ([]byte, error) {
	zerolog.Ctx(ctx).Debug().
		Str("entity", entity).
		Str("linked_entity", linkedEntity).
		Str("linked_id", linkedID).
		Int("limit", limit).
		Int("offset", offset).
		Msg("provider browse")

	params := url.Values{}
	params.Set(linkedEntity, linkedID)
	params.Set("limit", fmt.Sprintf("%d", limit))
	params.Set("offset", fmt.Sprintf("%d", offset))

	if joined := joinIncludes(inc); joined != "" {
		params.Set("inc", joined)
	}

	return c.get(ctx, entity, params)
}

func (c *client) search(ctx context.Context, entity, query string, limit, offset int) ([]byte, error) {
	zerolog.Ctx(ctx).Debug().Str("entity", entity).Str("query", query).Int("limit", limit).Int("offset", offset).Msg("provider search")

	params := url.Values{}
	params.Set("query", query)
	params.Set("limit", fmt.Sprintf("%d", limit))
	params.Set("offset", fmt.Sprintf("%d", offset))

	return c.get(ctx, entity, params)
}

func joinIncludes(inc []string) string {
	if len(inc) == 0 {
		return ""
	}
	return strings.Join(inc, "+")
}

func extractErrorMessage(body []byte) string {
	var errResp struct {
		Error string `json:"error"`
	}
	if json.Unmarshal(body, &errResp) == nil && errResp.Error != "" {
		return errResp.Error
	}
	return string(body)
}

func decode[T any](data []byte) (*T, error) {
	var result T
	if err := json.Unmarshal(data, &result); err != nil {
		return nil, fmt.Errorf("decode: %w", err)
	}
	return &result, nil
}