weatherreporter/internal/adapters/distributor/client.go

// Package distributor adapts weatherreporter report artifacts to distributor uploads.
package distributor

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"os"
	"strings"
	"time"

	distributorbundle "gitea.maximumdirect.net/eric/distributor/pkg/bundle"
	distributorupload "gitea.maximumdirect.net/eric/distributor/pkg/upload"

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

type Client struct {
	Endpoint        string
	TokenEnv        string
	Timeout         time.Duration
	newUploadClient uploadClientFactory
}

type UploadRequest struct {
	BundleID       string
	IdempotencyKey string
	SourcePath     string
	BundlePath     string
}

type UploadResult struct {
	RunID  string
	Status string
}

type IdempotencyConflictError struct {
	Err error
}

func (e *IdempotencyConflictError) Error() string {
	if e == nil || e.Err == nil {
		return "distributor idempotency conflict"
	}
	return e.Err.Error()
}

func (e *IdempotencyConflictError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.Err
}

type uploadClientFactory func(endpoint, token string, timeout time.Duration) (uploadClient, error)

type uploadClient interface {
	UploadFiles(ctx context.Context, opts uploadFilesOptions) (uploadFilesResult, error)
}

type uploadFilesOptions struct {
	BundleID       string
	IdempotencyKey string
	SourcePath     string
	BundlePath     string
}

type uploadFilesResult struct {
	RunID  string
	Status string
}

func New(cfg config.DistributorNotifyConfig) *Client {
	return newClient(cfg, newDistributorUploadClient)
}

func newClient(cfg config.DistributorNotifyConfig, factory uploadClientFactory) *Client {
	if factory == nil {
		factory = newDistributorUploadClient
	}
	return &Client{
		Endpoint:        cfg.Endpoint,
		TokenEnv:        cfg.TokenEnv,
		Timeout:         cfg.Timeout,
		newUploadClient: factory,
	}
}

func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, error) {
	if c == nil {
		return UploadResult{}, fmt.Errorf("distributor client is nil")
	}
	if c.Endpoint == "" {
		return UploadResult{}, fmt.Errorf("distributor endpoint is required")
	}
	if c.TokenEnv == "" {
		return UploadResult{}, fmt.Errorf("distributor token environment variable is required")
	}
	if req.BundleID == "" {
		return UploadResult{}, fmt.Errorf("distributor bundle id is required")
	}
	if req.IdempotencyKey == "" {
		return UploadResult{}, fmt.Errorf("distributor idempotency key is required for bundle %q", req.BundleID)
	}
	if req.SourcePath == "" {
		return UploadResult{}, fmt.Errorf("distributor source path is required for bundle %q", req.BundleID)
	}
	if req.BundlePath == "" {
		return UploadResult{}, fmt.Errorf("distributor bundle path is required for bundle %q", req.BundleID)
	}
	if c.newUploadClient == nil {
		return UploadResult{}, fmt.Errorf("distributor upload client factory is required for endpoint %q", c.Endpoint)
	}

	token := os.Getenv(c.TokenEnv)
	if token == "" {
		return UploadResult{}, fmt.Errorf("distributor token environment variable %q is not set", c.TokenEnv)
	}

	uploadClient, err := c.newUploadClient(c.Endpoint, token, c.Timeout)
	if err != nil {
		return UploadResult{}, fmt.Errorf("create distributor upload client for endpoint %q: %w", c.Endpoint, redactToken(err, token))
	}

	runCtx := ctx
	if runCtx == nil {
		runCtx = context.Background()
	}
	cancel := func() {}
	if c.Timeout > 0 {
		runCtx, cancel = context.WithTimeout(runCtx, c.Timeout)
	}
	defer cancel()

	result, err := uploadClient.UploadFiles(runCtx, uploadFilesOptions{
		BundleID:       req.BundleID,
		IdempotencyKey: req.IdempotencyKey,
		SourcePath:     req.SourcePath,
		BundlePath:     req.BundlePath,
	})
	if err != nil {
		return UploadResult{}, wrapUploadError(err, uploadErrorContext{
			Endpoint:       c.Endpoint,
			BundleID:       req.BundleID,
			IdempotencyKey: req.IdempotencyKey,
			SourcePath:     req.SourcePath,
			BundlePath:     req.BundlePath,
			Token:          token,
		})
	}

	return UploadResult{
		RunID:  result.RunID,
		Status: result.Status,
	}, nil
}

type distributorUploadClient struct {
	client *distributorupload.Client
}

func newDistributorUploadClient(endpoint, token string, timeout time.Duration) (uploadClient, error) {
	httpClient := (*http.Client)(nil)
	if timeout > 0 {
		httpClient = &http.Client{Timeout: timeout}
	}
	client, err := distributorupload.NewClient(distributorupload.ClientOptions{
		Endpoint:   endpoint,
		Token:      token,
		HTTPClient: httpClient,
	})
	if err != nil {
		return nil, err
	}
	return distributorUploadClient{client: client}, nil
}

func (c distributorUploadClient) UploadFiles(ctx context.Context, opts uploadFilesOptions) (uploadFilesResult, error) {
	result, err := c.client.UploadFiles(ctx, distributorupload.UploadFilesOptions{
		ID:             opts.BundleID,
		IdempotencyKey: opts.IdempotencyKey,
		Files: []distributorbundle.BundleFile{
			{SourcePath: opts.SourcePath, Path: opts.BundlePath},
		},
	})
	if err != nil {
		return uploadFilesResult{}, err
	}
	return uploadFilesResult{
		RunID:  result.RunID,
		Status: result.Status,
	}, nil
}

type uploadErrorContext struct {
	Endpoint       string
	BundleID       string
	IdempotencyKey string
	SourcePath     string
	BundlePath     string
	Token          string
}

func wrapUploadError(err error, ctx uploadErrorContext) error {
	var conflict *distributorupload.IdempotencyConflictError
	isConflict := errors.As(err, &conflict)
	err = redactToken(err, ctx.Token)
	if isConflict {
		return &IdempotencyConflictError{
			Err: fmt.Errorf("upload distributor bundle %q to endpoint %q with idempotency key %q from source %q as bundle path %q: idempotency conflict: %w", ctx.BundleID, ctx.Endpoint, ctx.IdempotencyKey, ctx.SourcePath, ctx.BundlePath, err),
		}
	}
	return fmt.Errorf("upload distributor bundle %q to endpoint %q with idempotency key %q from source %q as bundle path %q: %w", ctx.BundleID, ctx.Endpoint, ctx.IdempotencyKey, ctx.SourcePath, ctx.BundlePath, err)
}

func redactToken(err error, token string) error {
	if err == nil || token == "" {
		return err
	}
	return errors.New(redactTokenString(err.Error(), token))
}

func redactTokenString(value, token string) string {
	if token == "" {
		return value
	}
	return strings.ReplaceAll(value, token, "[redacted]")
}