Add producer upload client package

2026-06-04 14:10:51 +00:00
parent a15722571f
commit d637949db4
4 changed files with 1181 additions and 0 deletions
--- a/pkg/upload/client.go
+++ b/pkg/upload/client.go
@@ -0,0 +1,426 @@
+package upload
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"path"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/pkg/bundle"
+)
+
+const (
+	uploadPath            = "upload"
+	runsPath              = "runs"
+	idempotencyKeyHeader  = "Idempotency-Key"
+	defaultHTTPTimeout    = 30 * time.Second
+	defaultRetryAttempts  = 3
+	defaultRetryBaseDelay = 100 * time.Millisecond
+	defaultRetryMaxDelay  = time.Second
+	uploadContentTypeGzip = "application/gzip"
+	authorizationPrefix   = "Bearer "
+	redactedSecret        = "[redacted]"
+)
+
+func NewClient(opts ClientOptions) (*Client, error) {
+	endpoint, err := cleanEndpoint(opts.Endpoint)
+	if err != nil {
+		return nil, err
+	}
+	if opts.Token == "" {
+		return nil, fmt.Errorf("token is required")
+	}
+	retry, err := cleanRetryOptions(opts.Retry)
+	if err != nil {
+		return nil, err
+	}
+	httpClient := opts.HTTPClient
+	if httpClient == nil {
+		httpClient = &http.Client{Timeout: defaultHTTPTimeout}
+	}
+	return &Client{
+		endpoint:   endpoint,
+		token:      opts.Token,
+		httpClient: httpClient,
+		retry:      retry,
+	}, nil
+}
+
+func (c *Client) UploadBundle(ctx context.Context, opts UploadBundleOptions) (Result, error) {
+	if c == nil {
+		return Result{}, fmt.Errorf("client is nil")
+	}
+	if opts.Validate && opts.DisableValidation {
+		return Result{}, fmt.Errorf("validate and disable validation cannot both be set")
+	}
+	if opts.Root == "" {
+		return Result{}, fmt.Errorf("root is required")
+	}
+	manifest, err := bundle.LoadManifest(opts.Root)
+	if err != nil {
+		return Result{}, c.redactError(err)
+	}
+	if shouldValidateBundle(opts.Validate, opts.DisableValidation) {
+		if err := bundle.ValidateBundle(opts.Root, manifest); err != nil {
+			return Result{}, c.redactError(err)
+		}
+	}
+	archive, err := archiveBundle(opts.Root, manifest)
+	if err != nil {
+		return Result{}, c.redactError(err)
+	}
+	key, err := uploadIdempotencyKey(opts.IdempotencyKey)
+	if err != nil {
+		return Result{}, err
+	}
+	return c.uploadArchive(ctx, archive, key)
+}
+
+func (c *Client) UploadFiles(ctx context.Context, opts UploadFilesOptions) (Result, error) {
+	if c == nil {
+		return Result{}, fmt.Errorf("client is nil")
+	}
+	if opts.Validate && opts.DisableValidation {
+		return Result{}, fmt.Errorf("validate and disable validation cannot both be set")
+	}
+	if opts.ID == "" {
+		return Result{}, fmt.Errorf("id is required")
+	}
+	if len(opts.Files) == 0 {
+		return Result{}, fmt.Errorf("files is required")
+	}
+	tempRoot, err := os.MkdirTemp(opts.TempDir, "distributor-upload-*")
+	if err != nil {
+		return Result{}, c.redactError(fmt.Errorf("create temporary bundle root: %w", err))
+	}
+	defer func() {
+		_ = os.RemoveAll(tempRoot)
+	}()
+	localBundleRoot := filepath.Join(tempRoot, "bundle")
+	manifest, err := bundle.WriteBundle(bundle.WriteBundleOptions{
+		Root:    localBundleRoot,
+		ID:      opts.ID,
+		Created: opts.Created,
+		Files:   opts.Files,
+	})
+	if err != nil {
+		return Result{}, c.redactError(err)
+	}
+	if shouldValidateBundle(opts.Validate, opts.DisableValidation) {
+		if err := bundle.ValidateBundle(localBundleRoot, manifest); err != nil {
+			return Result{}, c.redactError(err)
+		}
+	}
+	archive, err := archiveBundle(localBundleRoot, manifest)
+	if err != nil {
+		return Result{}, c.redactError(err)
+	}
+	key, err := uploadIdempotencyKey(opts.IdempotencyKey)
+	if err != nil {
+		return Result{}, err
+	}
+	return c.uploadArchive(ctx, archive, key)
+}
+
+func (c *Client) Status(ctx context.Context, runID string) (RunStatus, error) {
+	if c == nil {
+		return RunStatus{}, fmt.Errorf("client is nil")
+	}
+	if runID == "" {
+		return RunStatus{}, fmt.Errorf("run id is required")
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if err := ctx.Err(); err != nil {
+		return RunStatus{}, err
+	}
+	request, err := http.NewRequestWithContext(ctx, http.MethodGet, c.statusURL(runID), nil)
+	if err != nil {
+		return RunStatus{}, c.redactError(err)
+	}
+	c.authorize(request)
+	response, err := c.httpClient.Do(request)
+	if err != nil {
+		return RunStatus{}, c.redactError(err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		return RunStatus{}, c.responseError(response)
+	}
+	var status RunStatus
+	if err := json.NewDecoder(response.Body).Decode(&status); err != nil {
+		return RunStatus{}, c.redactError(fmt.Errorf("decode run status: %w", err))
+	}
+	return status, nil
+}
+
+func (c *Client) uploadArchive(ctx context.Context, archive []byte, idempotencyKey string) (Result, error) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	var lastErr error
+	for attempt := 1; attempt <= c.retry.MaxAttempts; attempt++ {
+		if err := ctx.Err(); err != nil {
+			return Result{}, err
+		}
+		result, retry, err := c.uploadAttempt(ctx, archive, idempotencyKey)
+		if err == nil {
+			return result, nil
+		}
+		lastErr = err
+		if !retry || attempt == c.retry.MaxAttempts {
+			return Result{}, err
+		}
+		if err := waitForRetry(ctx, retryDelay(c.retry, attempt)); err != nil {
+			return Result{}, err
+		}
+	}
+	return Result{}, lastErr
+}
+
+func (c *Client) uploadAttempt(ctx context.Context, archive []byte, idempotencyKey string) (Result, bool, error) {
+	request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.uploadURL(), bytes.NewReader(archive))
+	if err != nil {
+		return Result{}, false, c.redactError(err)
+	}
+	c.authorize(request)
+	request.Header.Set("Content-Type", uploadContentTypeGzip)
+	request.Header.Set(idempotencyKeyHeader, idempotencyKey)
+
+	response, err := c.httpClient.Do(request)
+	if err != nil {
+		if ctxErr := ctx.Err(); ctxErr != nil {
+			return Result{}, false, ctxErr
+		}
+		return Result{}, isRetryableNetworkError(err), c.redactError(err)
+	}
+	defer response.Body.Close()
+
+	if response.StatusCode == http.StatusAccepted {
+		var result Result
+		if err := json.NewDecoder(response.Body).Decode(&result); err != nil {
+			return Result{}, false, c.redactError(fmt.Errorf("decode upload response: %w", err))
+		}
+		if result.RunID == "" {
+			return Result{}, false, fmt.Errorf("upload response run_id is required")
+		}
+		return result, false, nil
+	}
+	err = c.responseError(response)
+	return Result{}, response.StatusCode == http.StatusServiceUnavailable, err
+}
+
+func (c *Client) authorize(request *http.Request) {
+	request.Header.Set("Authorization", authorizationPrefix+c.token)
+}
+
+func (c *Client) uploadURL() string {
+	return joinEndpointPath(c.endpoint, uploadPath)
+}
+
+func (c *Client) statusURL(runID string) string {
+	return joinEndpointPath(c.endpoint, runsPath, runID)
+}
+
+func (c *Client) responseError(response *http.Response) error {
+	body, readErr := io.ReadAll(response.Body)
+	message := http.StatusText(response.StatusCode)
+	retryable := false
+	if readErr == nil && len(body) > 0 {
+		var decoded struct {
+			Error     string `json:"error"`
+			Retryable bool   `json:"retryable"`
+		}
+		if err := json.Unmarshal(body, &decoded); err == nil && decoded.Error != "" {
+			message = decoded.Error
+			retryable = decoded.Retryable
+		} else if trimmed := strings.TrimSpace(string(body)); trimmed != "" {
+			message = trimmed
+		}
+	}
+	message = c.redactString(message)
+	status := c.redactString(response.Status)
+	httpErr := HTTPError{
+		StatusCode: response.StatusCode,
+		Status:     status,
+		Message:    message,
+		Retryable:  retryable,
+	}
+	if response.StatusCode == http.StatusConflict {
+		return &IdempotencyConflictError{HTTPError: httpErr}
+	}
+	return &httpErr
+}
+
+func (c *Client) redactError(err error) error {
+	if err == nil {
+		return nil
+	}
+	message := c.redactString(err.Error())
+	if message == err.Error() {
+		return err
+	}
+	return errors.New(message)
+}
+
+func (c *Client) redactString(value string) string {
+	if c == nil || c.token == "" {
+		return value
+	}
+	return strings.ReplaceAll(value, c.token, redactedSecret)
+}
+
+func cleanEndpoint(value string) (string, error) {
+	if value == "" {
+		return "", fmt.Errorf("endpoint is required")
+	}
+	parsed, err := url.Parse(value)
+	if err != nil {
+		return "", fmt.Errorf("endpoint is invalid: %w", err)
+	}
+	if parsed.Scheme != "http" && parsed.Scheme != "https" {
+		return "", fmt.Errorf("endpoint scheme must be http or https")
+	}
+	if parsed.Host == "" {
+		return "", fmt.Errorf("endpoint host is required")
+	}
+	if parsed.User != nil {
+		return "", fmt.Errorf("endpoint userinfo is not supported")
+	}
+	if parsed.RawQuery != "" || parsed.Fragment != "" {
+		return "", fmt.Errorf("endpoint must not include query or fragment")
+	}
+	parsed.Path = strings.TrimRight(parsed.Path, "/")
+	parsed.RawPath = ""
+	return parsed.String(), nil
+}
+
+func cleanRetryOptions(opts RetryOptions) (RetryOptions, error) {
+	if opts.MaxAttempts < 0 {
+		return RetryOptions{}, fmt.Errorf("retry max attempts must be non-negative")
+	}
+	if opts.BaseDelay < 0 {
+		return RetryOptions{}, fmt.Errorf("retry base delay must be non-negative")
+	}
+	if opts.MaxDelay < 0 {
+		return RetryOptions{}, fmt.Errorf("retry max delay must be non-negative")
+	}
+	if opts.MaxAttempts == 0 {
+		opts.MaxAttempts = defaultRetryAttempts
+	}
+	if opts.BaseDelay == 0 {
+		opts.BaseDelay = defaultRetryBaseDelay
+	}
+	if opts.MaxDelay == 0 {
+		opts.MaxDelay = defaultRetryMaxDelay
+	}
+	if opts.MaxDelay < opts.BaseDelay {
+		return RetryOptions{}, fmt.Errorf("retry max delay must be greater than or equal to base delay")
+	}
+	return opts, nil
+}
+
+func uploadIdempotencyKey(value string) (string, error) {
+	if value == "" {
+		return randomIdempotencyKey()
+	}
+	if err := validateIdempotencyKey(value); err != nil {
+		return "", err
+	}
+	return value, nil
+}
+
+func validateIdempotencyKey(value string) error {
+	if value == "" {
+		return fmt.Errorf("idempotency key is required")
+	}
+	if len(value) > 128 {
+		return fmt.Errorf("idempotency key must be at most 128 bytes")
+	}
+	for index := 0; index < len(value); index++ {
+		character := value[index]
+		if character >= 'a' && character <= 'z' ||
+			character >= 'A' && character <= 'Z' ||
+			character >= '0' && character <= '9' ||
+			character == '.' ||
+			character == '_' ||
+			character == '-' ||
+			character == ':' {
+			continue
+		}
+		return fmt.Errorf("idempotency key contains unsupported character")
+	}
+	return nil
+}
+
+func randomIdempotencyKey() (string, error) {
+	var data [16]byte
+	if _, err := rand.Read(data[:]); err != nil {
+		return "", fmt.Errorf("generate idempotency key: %w", err)
+	}
+	return hex.EncodeToString(data[:]), nil
+}
+
+func shouldValidateBundle(validate, disable bool) bool {
+	return validate || !disable
+}
+
+func retryDelay(opts RetryOptions, attempt int) time.Duration {
+	delay := opts.BaseDelay
+	for index := 1; index < attempt; index++ {
+		delay *= 2
+		if delay >= opts.MaxDelay {
+			return opts.MaxDelay
+		}
+	}
+	return delay
+}
+
+func waitForRetry(ctx context.Context, delay time.Duration) error {
+	timer := time.NewTimer(delay)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
+func isRetryableNetworkError(err error) bool {
+	if err == nil {
+		return false
+	}
+	var netErr net.Error
+	if errors.As(err, &netErr) {
+		return netErr.Timeout() || netErr.Temporary()
+	}
+	return true
+}
+
+func joinEndpointPath(endpoint string, elements ...string) string {
+	parsed, err := url.Parse(endpoint)
+	if err != nil {
+		return endpoint
+	}
+	parts := []string{}
+	if parsed.Path != "" && parsed.Path != "/" {
+		parts = append(parts, strings.Trim(parsed.Path, "/"))
+	}
+	parts = append(parts, elements...)
+	parsed.Path = "/" + path.Join(parts...)
+	return parsed.String()
+}