From ce43a6044a8b523965037af35f41f2f7e4b60fc7 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 8 Jun 2026 04:37:33 +0000 Subject: [PATCH] Route upload client by pipeline --- pkg/upload/client.go | 36 +++++++++---- pkg/upload/client_test.go | 104 ++++++++++++++++++++++++++++++++++---- pkg/upload/doc.go | 29 +++++++---- pkg/upload/types.go | 2 + 4 files changed, 142 insertions(+), 29 deletions(-) diff --git a/pkg/upload/client.go b/pkg/upload/client.go index 93ce680..aef866b 100644 --- a/pkg/upload/client.go +++ b/pkg/upload/client.go @@ -15,6 +15,7 @@ import ( "os" "path" "path/filepath" + "regexp" "strings" "time" @@ -22,7 +23,6 @@ import ( ) const ( - uploadPath = "upload" runsPath = "runs" idempotencyKeyHeader = "Idempotency-Key" defaultHTTPTimeout = 30 * time.Second @@ -34,6 +34,8 @@ const ( redactedSecret = "[redacted]" ) +var pipelineIDPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._-]*$`) + func NewClient(opts ClientOptions) (*Client, error) { endpoint, err := cleanEndpoint(opts.Endpoint) if err != nil { @@ -65,6 +67,9 @@ func (c *Client) UploadBundle(ctx context.Context, opts UploadBundleOptions) (Re if opts.Validate && opts.DisableValidation { return Result{}, fmt.Errorf("validate and disable validation cannot both be set") } + if err := validatePipelineID(opts.PipelineID); err != nil { + return Result{}, err + } if opts.Root == "" { return Result{}, fmt.Errorf("root is required") } @@ -85,7 +90,7 @@ func (c *Client) UploadBundle(ctx context.Context, opts UploadBundleOptions) (Re if err != nil { return Result{}, err } - return c.uploadArchive(ctx, archive, key) + return c.uploadArchive(ctx, opts.PipelineID, archive, key) } func (c *Client) UploadFiles(ctx context.Context, opts UploadFilesOptions) (Result, error) { @@ -95,6 +100,9 @@ func (c *Client) UploadFiles(ctx context.Context, opts UploadFilesOptions) (Resu if opts.Validate && opts.DisableValidation { return Result{}, fmt.Errorf("validate and disable validation cannot both be set") } + if err := validatePipelineID(opts.PipelineID); err != nil { + return Result{}, err + } if opts.ID == "" { return Result{}, fmt.Errorf("id is required") } @@ -131,7 +139,7 @@ func (c *Client) UploadFiles(ctx context.Context, opts UploadFilesOptions) (Resu if err != nil { return Result{}, err } - return c.uploadArchive(ctx, archive, key) + return c.uploadArchive(ctx, opts.PipelineID, archive, key) } func (c *Client) Status(ctx context.Context, runID string) (RunStatus, error) { @@ -167,7 +175,7 @@ func (c *Client) Status(ctx context.Context, runID string) (RunStatus, error) { return status, nil } -func (c *Client) uploadArchive(ctx context.Context, archive []byte, idempotencyKey string) (Result, error) { +func (c *Client) uploadArchive(ctx context.Context, pipelineID string, archive []byte, idempotencyKey string) (Result, error) { if ctx == nil { ctx = context.Background() } @@ -176,7 +184,7 @@ func (c *Client) uploadArchive(ctx context.Context, archive []byte, idempotencyK if err := ctx.Err(); err != nil { return Result{}, err } - result, retry, err := c.uploadAttempt(ctx, archive, idempotencyKey) + result, retry, err := c.uploadAttempt(ctx, pipelineID, archive, idempotencyKey) if err == nil { return result, nil } @@ -191,8 +199,8 @@ func (c *Client) uploadArchive(ctx context.Context, archive []byte, idempotencyK 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)) +func (c *Client) uploadAttempt(ctx context.Context, pipelineID string, archive []byte, idempotencyKey string) (Result, bool, error) { + request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.uploadURL(pipelineID), bytes.NewReader(archive)) if err != nil { return Result{}, false, c.redactError(err) } @@ -227,8 +235,8 @@ 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) uploadURL(pipelineID string) string { + return joinEndpointPath(c.endpoint, "v1", "pipelines", pipelineID, "upload") } func (c *Client) statusURL(runID string) string { @@ -343,6 +351,16 @@ func uploadIdempotencyKey(value string) (string, error) { return value, nil } +func validatePipelineID(value string) error { + if value == "" { + return fmt.Errorf("pipeline id is required") + } + if !pipelineIDPattern.MatchString(value) { + return fmt.Errorf("pipeline id must be a slug-like identifier") + } + return nil +} + func validateIdempotencyKey(value string) error { if value == "" { return fmt.Errorf("idempotency key is required") diff --git a/pkg/upload/client_test.go b/pkg/upload/client_test.go index 5dd6349..a218596 100644 --- a/pkg/upload/client_test.go +++ b/pkg/upload/client_test.go @@ -47,7 +47,7 @@ func TestNewClientValidatesOptions(t *testing.T) { if err != nil { t.Fatalf("NewClient() error = %v", err) } - if got, want := client.uploadURL(), "http://127.0.0.1:8080/base/upload"; got != want { + if got, want := client.uploadURL("reports.daily"), "http://127.0.0.1:8080/base/v1/pipelines/reports.daily/upload"; got != want { t.Fatalf("upload URL = %q, want %q", got, want) } if client.httpClient == nil || client.httpClient.Timeout == 0 { @@ -65,7 +65,7 @@ func TestUploadBundleSendsCallerKeyAndManifestArchive(t *testing.T) { } server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if got, want := r.URL.Path, "/upload"; got != want { + if got, want := r.URL.Path, "/v1/pipelines/reports.daily/upload"; got != want { t.Fatalf("path = %q, want %q", got, want) } if got, want := r.Header.Get("Authorization"), "Bearer secret-token"; got != want { @@ -93,6 +93,7 @@ func TestUploadBundleSendsCallerKeyAndManifestArchive(t *testing.T) { t.Fatalf("NewClient() error = %v", err) } result, err := client.UploadBundle(context.Background(), UploadBundleOptions{ + PipelineID: "reports.daily", Root: root, IdempotencyKey: "producer.retry:one", }) @@ -112,6 +113,9 @@ func TestUploadFilesBuildsTemporaryBundleWithoutTouchingSources(t *testing.T) { } tempDir := t.TempDir() server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got, want := r.URL.Path, "/v1/pipelines/reports.files/upload"; got != want { + t.Fatalf("path = %q, want %q", got, want) + } entries := readArchiveEntries(t, r.Body) if got := string(entries["manifest.json"]); !strings.Contains(got, `"id": "reports.from.files"`) { t.Fatalf("manifest = %s, want uploaded id", got) @@ -131,7 +135,8 @@ func TestUploadFilesBuildsTemporaryBundleWithoutTouchingSources(t *testing.T) { t.Fatalf("NewClient() error = %v", err) } _, err = client.UploadFiles(context.Background(), UploadFilesOptions{ - ID: "reports.from.files", + PipelineID: "reports.files", + ID: "reports.from.files", Files: []sourcebundle.BundleFile{{ SourcePath: sourcePath, Path: "reports/report.md", @@ -153,6 +158,77 @@ func TestUploadFilesBuildsTemporaryBundleWithoutTouchingSources(t *testing.T) { } } +func TestUploadMethodsRequirePipelineIDBeforeLocalWork(t *testing.T) { + var requests atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + t.Fatal("server should not receive request") + })) + defer server.Close() + + client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()}) + if err != nil { + t.Fatalf("NewClient() error = %v", err) + } + + missingRoot := filepath.Join(t.TempDir(), "missing") + if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: missingRoot}); err == nil || !strings.Contains(err.Error(), "pipeline id is required") { + t.Fatalf("UploadBundle() error = %v, want missing pipeline id", err) + } + + tempDir := t.TempDir() + sourcePath := filepath.Join(t.TempDir(), "report.md") + if err := os.WriteFile(sourcePath, []byte("data"), 0o600); err != nil { + t.Fatalf("write source: %v", err) + } + if _, err := client.UploadFiles(context.Background(), UploadFilesOptions{ + ID: "reports.from.files", + Files: []sourcebundle.BundleFile{{ + SourcePath: sourcePath, + Path: "report.md", + }}, + TempDir: tempDir, + }); err == nil || !strings.Contains(err.Error(), "pipeline id is required") { + t.Fatalf("UploadFiles() error = %v, want missing pipeline id", err) + } + entries, err := os.ReadDir(tempDir) + if err != nil { + t.Fatalf("read temp dir: %v", err) + } + if len(entries) != 0 { + t.Fatalf("temp dir entries = %d, want no local bundle work", len(entries)) + } + if got := requests.Load(); got != 0 { + t.Fatalf("requests = %d, want 0", got) + } +} + +func TestUploadMethodsRejectInvalidPipelineIDBeforeHTTPRequest(t *testing.T) { + root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}}) + var requests atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + t.Fatal("server should not receive request") + })) + defer server.Close() + + client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()}) + if err != nil { + t.Fatalf("NewClient() error = %v", err) + } + for _, pipelineID := range []string{".reports", "reports/daily", "reports daily"} { + t.Run(pipelineID, func(t *testing.T) { + _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: pipelineID, Root: root}) + if err == nil || !strings.Contains(err.Error(), "pipeline id must be a slug-like identifier") { + t.Fatalf("UploadBundle() error = %v, want invalid pipeline id", err) + } + }) + } + if got := requests.Load(); got != 0 { + t.Fatalf("requests = %d, want 0", got) + } +} + func TestUploadBundleValidationFailurePreventsHTTPRequest(t *testing.T) { root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "original"}}) if err := os.WriteFile(filepath.Join(root, "report.md"), []byte("changed"), 0o600); err != nil { @@ -169,7 +245,7 @@ func TestUploadBundleValidationFailurePreventsHTTPRequest(t *testing.T) { if err != nil { t.Fatalf("NewClient() error = %v", err) } - if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root}); err == nil { + if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root}); err == nil { t.Fatal("UploadBundle() error = nil, want validation error") } if got := requests.Load(); got != 0 { @@ -193,7 +269,7 @@ func TestUploadBundleCanDisableLocalValidation(t *testing.T) { if err != nil { t.Fatalf("NewClient() error = %v", err) } - if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, DisableValidation: true}); err != nil { + if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, DisableValidation: true}); err != nil { t.Fatalf("UploadBundle() error = %v", err) } if got := requests.Load(); got != 1 { @@ -206,6 +282,9 @@ func TestGeneratedIdempotencyKeyIsReusedAcrossRetry(t *testing.T) { var attempts atomic.Int64 var keys []string server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got, want := r.URL.Path, "/v1/pipelines/reports/upload"; got != want { + t.Fatalf("path = %q, want %q", got, want) + } keys = append(keys, r.Header.Get(idempotencyKeyHeader)) if attempts.Add(1) == 1 { writeJSONError(w, http.StatusServiceUnavailable, "busy", false) @@ -224,7 +303,7 @@ func TestGeneratedIdempotencyKeyIsReusedAcrossRetry(t *testing.T) { if err != nil { t.Fatalf("NewClient() error = %v", err) } - if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root}); err != nil { + if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root}); err != nil { t.Fatalf("UploadBundle() error = %v", err) } if got, want := attempts.Load(), int64(2); got != want { @@ -276,7 +355,7 @@ func TestUploadResponseParsingAndNoRetryStatuses(t *testing.T) { if err != nil { t.Fatalf("NewClient() error = %v", err) } - _, err = client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "key"}) + _, err = client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "key"}) if err == nil { t.Fatal("UploadBundle() error = nil, want error") } @@ -309,7 +388,7 @@ func TestTokenRedactedFromHTTPError(t *testing.T) { if err != nil { t.Fatalf("NewClient() error = %v", err) } - _, err = client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "key"}) + _, err = client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "key"}) if err == nil { t.Fatal("UploadBundle() error = nil, want error") } @@ -330,6 +409,9 @@ func TestNetworkRetryUsesSameIdempotencyKey(t *testing.T) { Token: "secret", HTTPClient: &http.Client{Transport: roundTripFunc(func(request *http.Request) (*http.Response, error) { keys = append(keys, request.Header.Get(idempotencyKeyHeader)) + if got, want := request.URL.Path, "/v1/pipelines/reports/upload"; got != want { + t.Fatalf("path = %q, want %q", got, want) + } if attempts.Add(1) == 1 { return nil, temporaryNetworkError{} } @@ -346,7 +428,7 @@ func TestNetworkRetryUsesSameIdempotencyKey(t *testing.T) { if err != nil { t.Fatalf("NewClient() error = %v", err) } - result, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "network-retry"}) + result, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "network-retry"}) if err != nil { t.Fatalf("UploadBundle() error = %v", err) } @@ -381,7 +463,7 @@ func TestContextCancellationDuringRetryBackoff(t *testing.T) { if err != nil { t.Fatalf("NewClient() error = %v", err) } - _, err = client.UploadBundle(ctx, UploadBundleOptions{Root: root, IdempotencyKey: "cancel"}) + _, err = client.UploadBundle(ctx, UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "cancel"}) if !errors.Is(err, context.Canceled) { t.Fatalf("UploadBundle() error = %v, want context.Canceled", err) } @@ -446,7 +528,7 @@ func TestInvalidCallerIdempotencyKeyPreventsHTTPRequest(t *testing.T) { if err != nil { t.Fatalf("NewClient() error = %v", err) } - if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "bad key"}); err == nil { + if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "bad key"}); err == nil { t.Fatal("UploadBundle() error = nil, want invalid key error") } if got := requests.Load(); got != 0 { diff --git a/pkg/upload/doc.go b/pkg/upload/doc.go index 69bdeb9..f0afbd8 100644 --- a/pkg/upload/doc.go +++ b/pkg/upload/doc.go @@ -12,9 +12,10 @@ // // NewClient creates a Client from ClientOptions. Endpoint is required and must // be an http or https distributor server base URL without userinfo, query, or -// fragment. The client derives /upload for submissions and /runs/ for -// status checks. Token is required and is sent as Authorization: Bearer . -// Token values are redacted from errors produced by the client. +// fragment. The client derives /v1/pipelines//upload for +// submissions and /runs/ for status checks. Token is required and is +// sent as Authorization: Bearer . Token values are redacted from errors +// produced by the client. // // HTTPClient is optional. When omitted, the package uses a client with a // conservative timeout. Retry is optional; zero values select safe defaults. @@ -23,20 +24,28 @@ // // # Upload Workflows // -// UploadBundle uploads an existing local source bundle root. The root must -// contain manifest.json. By default, UploadBundle loads the manifest and -// validates the complete local bundle with pkg/bundle before making any HTTP -// request. The generated gzip-compressed tar archive contains manifest.json and -// exactly the manifest-listed files; unlisted files are not uploaded. +// UploadBundle uploads an existing local source bundle root to the configured +// PipelineID. PipelineID is required and must match the server's slug-like +// pipeline id syntax. The root must contain manifest.json. By default, +// UploadBundle loads the manifest and validates the complete local bundle with +// pkg/bundle before making any HTTP request. The generated gzip-compressed tar +// archive contains manifest.json and exactly the manifest-listed files; +// unlisted files are not uploaded. // // UploadFiles is the convenience workflow for producer applications that have -// generated files but have not yet assembled a bundle directory. It uses +// generated files but have not yet assembled a bundle directory. PipelineID is +// required and selects the configured distributor workflow. UploadFiles uses // pkg/bundle to create a temporary complete bundle from explicit // bundle.BundleFile values, validates it by default, archives it, uploads it, // and removes temporary files when the call returns. UploadFiles does not write // into producer source directories. A zero Created timestamp follows // pkg/bundle defaulting behavior. // +// The producer contract has four separate identifiers: the bearer token +// authenticates the client, PipelineID selects the distributor workflow, the +// source manifest ID identifies the logical artifact within that workflow, and +// IdempotencyKey identifies one producer run and retry group. +// // Validation is enabled by default. Set DisableValidation when the application // has already performed equivalent local validation and wants to skip the // package's validation step. Validate and DisableValidation must not both be @@ -91,6 +100,7 @@ // } // // result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{ +// PipelineID: "reports.daily", // ID: "reports.daily.2026-06-06", // IdempotencyKey: "reports.daily.2026-06-06", // Files: []bundle.BundleFile{ @@ -115,6 +125,7 @@ // Example: upload an existing bundle root. // // result, err := client.UploadBundle(ctx, upload.UploadBundleOptions{ +// PipelineID: "reports.daily", // Root: "/var/lib/reports/daily-2026-06-06", // IdempotencyKey: "reports.daily.2026-06-06", // }) diff --git a/pkg/upload/types.go b/pkg/upload/types.go index db3b89a..4de6035 100644 --- a/pkg/upload/types.go +++ b/pkg/upload/types.go @@ -30,6 +30,7 @@ type RetryOptions struct { } type UploadBundleOptions struct { + PipelineID string Root string Validate bool DisableValidation bool @@ -37,6 +38,7 @@ type UploadBundleOptions struct { } type UploadFilesOptions struct { + PipelineID string ID string Created time.Time Files []bundle.BundleFile