Cleanup remotefile resolver and delete internal/worker

pkg/customizations/remotefile/client.go → pkg/remotefile/client.go

@@ -1,6 +1,7 @@
 package remotefile
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"net/http"
@@ -17,8 +18,8 @@ func NewClient() *Client {
 	}
 }
 
-func (c *Client) makeRequest(u *url.URL) ([]byte, error) {
-	req, err := http.NewRequest("GET", u.String(), nil)
+func (c *Client) makeRequest(ctx context.Context, u *url.URL) ([]byte, error) {
+	req, err := http.NewRequestWithContext(ctx, "GET", u.String(), nil)
 	if err != nil {
 		return nil, err
 	}
@@ -27,8 +28,12 @@ func (c *Client) makeRequest(u *url.URL) ([]byte, error) {
 	if err != nil {
 		return nil, err
 	}
-
 	defer resp.Body.Close()
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return nil, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, u.String())
+	}
+
 	output, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return nil, err
@@ -50,11 +55,11 @@ func (c *Client) validateURL(u string) (*url.URL, error) {
 
 // resolve and return the contents of a remote file
 // which can be used later, in the pipeline
-func (c *Client) Resolve(u string) ([]byte, error) {
+func (c *Client) Resolve(ctx context.Context, u string) ([]byte, error) {
 	parsedURL, err := c.validateURL(u)
 	if err != nil {
 		return nil, err
 	}
 
-	return c.makeRequest(parsedURL)
+	return c.makeRequest(ctx, parsedURL)
 }

pkg/customizations/remotefile/client_test.go → pkg/remotefile/client_test.go

@@ -1,6 +1,7 @@
 package remotefile
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
@@ -19,6 +20,9 @@ func makeTestServer() *httptest.Server {
 		if r.URL.Path == "/key2" {
 			fmt.Fprintln(w, "key2")
 		}
+		if r.URL.Path == "/notfound" {
+			w.WriteHeader(http.StatusNotFound)
+		}
 	}))
 }
 
@@ -29,14 +33,23 @@ func TestClientResolve(t *testing.T) {
 
 	client := NewClient()
 
-	output, err := client.Resolve(url)
+	output, err := client.Resolve(context.Background(), url)
 	assert.NoError(t, err)
 
 	expectedOutput := "key1\n"
 
 	assert.Equal(t, expectedOutput, string(output))
 }
 
+func TestClientResolveNonOKStatus(t *testing.T) {
+	server := makeTestServer()
+
+	client := NewClient()
+	_, err := client.Resolve(context.Background(), server.URL+"/notfound")
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "unexpected status 404")
+}
+
 func TestInputSpecValidation(t *testing.T) {
 	server := makeTestServer()
 

pkg/remotefile/resolver.go

@@ -0,0 +1,72 @@
+package remotefile
+
+import (
+	"context"
+	"fmt"
+	"slices"
+	"sort"
+	"strings"
+	"sync"
+)
+
+type resolveResult struct {
+	url     string
+	content []byte
+	err     error
+}
+
+// TODO: could make this more generic since this is shared with the container
+// resolver
+type Resolver struct {
+	queue chan resolveResult
+	wg    sync.WaitGroup
+	ctx   context.Context
+}
+
+func NewResolver(ctx context.Context) *Resolver {
+	return &Resolver{
+		queue: make(chan resolveResult),
+		wg:    sync.WaitGroup{},
+		ctx:   ctx,
+	}
+}
+
+// Add a URL to the resolver queue. When called after Finish was called,
+// it may panic.
+func (r *Resolver) Add(url string) {
+	r.wg.Add(1)
+	client := NewClient()
+
+	go func() {
+		defer r.wg.Done()
+
+		content, err := client.Resolve(r.ctx, url)
+		r.queue <- resolveResult{url: url, content: content, err: err}
+	}()
+}
+
+// Finish starts collecting of results and returns them. No further calls to Add
+// are allowed after this call. It blocks until all results are collected.
+func (r *Resolver) Finish() ([]Spec, error) {
+	go func() {
+		r.wg.Wait()
+		close(r.queue)
+	}()
+
+	var resultItems []Spec
+	var errs []string
+	for result := range r.queue {
+		if result.err == nil {
+			resultItems = append(resultItems, Spec{URL: result.url, Content: result.content})
+		} else {
+			errs = append(errs, result.err.Error())
+		}
+	}
+
+	if len(errs) > 0 {
+		sort.Strings(errs)
+		return resultItems, fmt.Errorf("failed to resolve remote files: %s", strings.Join(slices.Compact(errs), "; "))
+	}
+
+	return resultItems, nil
+}