eric/weatherreporter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Updated the distributor bundle path template

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-08 10:29:42 -05:00
parent d71c7e4d28
commit 8577fc29e4
18 changed files with 286 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
internal/adapters/distributor/client.go
View File

@@ -28,11 +28,15 @@ type UploadRequest struct {
PipelineID string
BundleID string
IdempotencyKey string
SourcePath string
BundlePath string
Files []UploadFile
CreatedAt time.Time
}
type UploadFile struct {
SourcePath string
BundlePath string
}
type UploadResult struct {
RunID string
Status string
@@ -81,8 +85,7 @@ type uploadFilesOptions struct {
PipelineID string
BundleID string
IdempotencyKey string
SourcePath string
BundlePath string
Files []UploadFile
CreatedAt time.Time
}
@@ -139,11 +142,16 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e
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 len(req.Files) == 0 {
return UploadResult{}, fmt.Errorf("distributor upload files are required for bundle %q", req.BundleID)
}
if req.BundlePath == "" {
return UploadResult{}, fmt.Errorf("distributor bundle path is required for bundle %q", req.BundleID)
for i, file := range req.Files {
if file.SourcePath == "" {
return UploadResult{}, fmt.Errorf("distributor source path is required for bundle %q file %d", req.BundleID, i)
}
if file.BundlePath == "" {
return UploadResult{}, fmt.Errorf("distributor bundle path is required for bundle %q file %d", req.BundleID, i)
}
}
if c.newUploadClient == nil {
return UploadResult{}, fmt.Errorf("distributor upload client factory is required for endpoint %q", c.Endpoint)
@@ -173,8 +181,7 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e
PipelineID: req.PipelineID,
BundleID: req.BundleID,
IdempotencyKey: req.IdempotencyKey,
SourcePath: req.SourcePath,
BundlePath: req.BundlePath,
Files: append([]UploadFile(nil), req.Files...),
CreatedAt: req.CreatedAt,
})
if err != nil {
@@ -183,8 +190,8 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e
PipelineID: req.PipelineID,
BundleID: req.BundleID,
IdempotencyKey: req.IdempotencyKey,
SourcePath: req.SourcePath,
BundlePath: req.BundlePath,
SourcePaths: uploadSourcePaths(req.Files),
BundlePaths: uploadBundlePaths(req.Files),
Token: token,
})
}
@@ -271,14 +278,19 @@ func newDistributorUploadClient(endpoint, token string, timeout time.Duration) (
}
func (c distributorUploadClient) UploadFiles(ctx context.Context, opts uploadFilesOptions) (uploadFilesResult, error) {
files := make([]distributorbundle.BundleFile, 0, len(opts.Files))
for _, file := range opts.Files {
files = append(files, distributorbundle.BundleFile{
SourcePath: file.SourcePath,
Path: file.BundlePath,
})
}
result, err := c.client.UploadFiles(ctx, distributorupload.UploadFilesOptions{
PipelineID: opts.PipelineID,
ID: opts.BundleID,
Created: opts.CreatedAt,
IdempotencyKey: opts.IdempotencyKey,
Files: []distributorbundle.BundleFile{
{SourcePath: opts.SourcePath, Path: opts.BundlePath},
},
Files: files,
})
if err != nil {
return uploadFilesResult{}, err
@@ -311,8 +323,8 @@ type uploadErrorContext struct {
PipelineID string
BundleID string
IdempotencyKey string
SourcePath string
BundlePath string
SourcePaths []string
BundlePaths []string
Token string
}
@@ -322,10 +334,26 @@ func wrapUploadError(err error, ctx uploadErrorContext) error {
err = redactToken(err, ctx.Token)
if isConflict {
return &IdempotencyConflictError{
Err: fmt.Errorf("upload distributor bundle %q to pipeline %q at endpoint %q with idempotency key %q from source %q as bundle path %q: idempotency conflict: %w", ctx.BundleID, ctx.PipelineID, ctx.Endpoint, ctx.IdempotencyKey, ctx.SourcePath, ctx.BundlePath, err),
Err: fmt.Errorf("upload distributor bundle %q to pipeline %q at endpoint %q with idempotency key %q from sources %q as bundle paths %q: idempotency conflict: %w", ctx.BundleID, ctx.PipelineID, ctx.Endpoint, ctx.IdempotencyKey, ctx.SourcePaths, ctx.BundlePaths, err),
}
}
return fmt.Errorf("upload distributor bundle %q to pipeline %q at endpoint %q with idempotency key %q from source %q as bundle path %q: %w", ctx.BundleID, ctx.PipelineID, ctx.Endpoint, ctx.IdempotencyKey, ctx.SourcePath, ctx.BundlePath, err)
return fmt.Errorf("upload distributor bundle %q to pipeline %q at endpoint %q with idempotency key %q from sources %q as bundle paths %q: %w", ctx.BundleID, ctx.PipelineID, ctx.Endpoint, ctx.IdempotencyKey, ctx.SourcePaths, ctx.BundlePaths, err)
}
func uploadSourcePaths(files []UploadFile) []string {
paths := make([]string, 0, len(files))
for _, file := range files {
paths = append(paths, file.SourcePath)
}
return paths
}
func uploadBundlePaths(files []UploadFile) []string {
paths := make([]string, 0, len(files))
for _, file := range files {
paths = append(paths, file.BundlePath)
}
return paths
}
func redactToken(err error, token string) error {

42
internal/adapters/distributor/client_test.go
View File

@@ -14,7 +14,7 @@ import (
"gitea.maximumdirect.net/eric/weatherreporter/internal/config"
)
func TestUploadUsesConfiguredClientAndSingleFile(t *testing.T) {
func TestUploadUsesConfiguredClientAndFiles(t *testing.T) {
cfg := config.Defaults().Notify.Distributor
cfg.Endpoint = "https://distributor.example.test"
cfg.TokenEnv = "DISTRIBUTOR_UPLOAD_TOKEN"
@@ -33,9 +33,11 @@ func TestUploadUsesConfiguredClientAndSingleFile(t *testing.T) {
PipelineID: "weatherreporter.daily",
BundleID: "weatherreporter.home.daily.run",
IdempotencyKey: "weatherreporter.home.daily.run",
SourcePath: "/tmp/report.md",
BundlePath: "daily.md",
CreatedAt: time.Date(2026, 6, 7, 12, 0, 0, 123, time.UTC),
Files: []UploadFile{
{SourcePath: "/tmp/report.md", BundlePath: "2026-06-07/daily/report.md"},
{SourcePath: "/tmp/report.md", BundlePath: "2026-06-07/daily/latest.md"},
},
CreatedAt: time.Date(2026, 6, 7, 12, 0, 0, 123, time.UTC),
})
if err != nil {
t.Fatalf("Upload() error = %v", err)
@@ -65,11 +67,14 @@ func TestUploadUsesConfiguredClientAndSingleFile(t *testing.T) {
if got.IdempotencyKey != "weatherreporter.home.daily.run" {
t.Fatalf("IdempotencyKey = %q, want weatherreporter.home.daily.run", got.IdempotencyKey)
}
if got.SourcePath != "/tmp/report.md" {
t.Fatalf("SourcePath = %q, want /tmp/report.md", got.SourcePath)
if len(got.Files) != 2 {
t.Fatalf("files = %#v, want two mappings", got.Files)
}
if got.BundlePath != "daily.md" {
t.Fatalf("BundlePath = %q, want daily.md", got.BundlePath)
if got.Files[0].SourcePath != "/tmp/report.md" || got.Files[0].BundlePath != "2026-06-07/daily/report.md" {
t.Fatalf("first file = %#v, want archive mapping", got.Files[0])
}
if got.Files[1].SourcePath != "/tmp/report.md" || got.Files[1].BundlePath != "2026-06-07/daily/latest.md" {
t.Fatalf("second file = %#v, want latest mapping", got.Files[1])
}
if got.CreatedAt.IsZero() {
t.Fatal("CreatedAt is zero, want generated report timestamp")
@@ -102,17 +107,24 @@ func TestUploadRejectsMissingInputs(t *testing.T) {
},
wantErr: "pipeline id is required",
},
{
name: "Files",
mutate: func(c *Client, req *UploadRequest) {
req.Files = nil
},
wantErr: "upload files are required",
},
{
name: "SourcePath",
mutate: func(c *Client, req *UploadRequest) {
req.SourcePath = ""
req.Files[0].SourcePath = ""
},
wantErr: "source path is required",
},
{
name: "BundlePath",
mutate: func(c *Client, req *UploadRequest) {
req.BundlePath = ""
req.Files[0].BundlePath = ""
},
wantErr: "bundle path is required",
},
@@ -181,7 +193,7 @@ func TestUploadWrapsUploadFailureWithContextWithoutToken(t *testing.T) {
if err == nil {
t.Fatal("Upload() error = nil, want error")
}
for _, want := range []string{cfg.Endpoint, req.PipelineID, req.BundleID, req.IdempotencyKey, req.SourcePath, req.BundlePath} {
for _, want := range []string{cfg.Endpoint, req.PipelineID, req.BundleID, req.IdempotencyKey, req.Files[0].SourcePath, req.Files[0].BundlePath, req.Files[1].BundlePath} {
if !strings.Contains(err.Error(), want) {
t.Fatalf("error = %q, want context %q", err.Error(), want)
}
@@ -332,9 +344,11 @@ func validUploadRequest() UploadRequest {
PipelineID: "weatherreporter.daily",
BundleID: "weatherreporter.home.daily.run",
IdempotencyKey: "weatherreporter.home.daily.run",
SourcePath: "/tmp/report.md",
BundlePath: "daily.md",
CreatedAt: time.Date(2026, 6, 7, 12, 0, 0, 123, time.UTC),
Files: []UploadFile{
{SourcePath: "/tmp/report.md", BundlePath: "2026-06-07/daily/report.md"},
{SourcePath: "/tmp/report.md", BundlePath: "2026-06-07/daily/latest.md"},
},
CreatedAt: time.Date(2026, 6, 7, 12, 0, 0, 123, time.UTC),
}
}