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

Implement the distributor v0.5 PipelineID update

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-08 07:04:31 -05:00
parent 9bc8156615
commit d71c7e4d28
24 changed files with 264 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -25,6 +25,7 @@ type Client struct {
}
type UploadRequest struct {
PipelineID string
BundleID string
IdempotencyKey string
SourcePath string
@@ -77,6 +78,7 @@ type uploadClient interface {
}
type uploadFilesOptions struct {
PipelineID string
BundleID string
IdempotencyKey string
SourcePath string
@@ -128,6 +130,9 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e
if c.TokenEnv == "" {
return UploadResult{}, fmt.Errorf("distributor token environment variable is required")
}
if req.PipelineID == "" {
return UploadResult{}, fmt.Errorf("distributor pipeline id is required")
}
if req.BundleID == "" {
return UploadResult{}, fmt.Errorf("distributor bundle id is required")
}
@@ -165,6 +170,7 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e
defer cancel()
result, err := uploadClient.UploadFiles(runCtx, uploadFilesOptions{
PipelineID: req.PipelineID,
BundleID: req.BundleID,
IdempotencyKey: req.IdempotencyKey,
SourcePath: req.SourcePath,
@@ -174,6 +180,7 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e
if err != nil {
return UploadResult{}, wrapUploadError(err, uploadErrorContext{
Endpoint: c.Endpoint,
PipelineID: req.PipelineID,
BundleID: req.BundleID,
IdempotencyKey: req.IdempotencyKey,
SourcePath: req.SourcePath,
@@ -265,6 +272,7 @@ func newDistributorUploadClient(endpoint, token string, timeout time.Duration) (
func (c distributorUploadClient) UploadFiles(ctx context.Context, opts uploadFilesOptions) (uploadFilesResult, error) {
result, err := c.client.UploadFiles(ctx, distributorupload.UploadFilesOptions{
PipelineID: opts.PipelineID,
ID: opts.BundleID,
Created: opts.CreatedAt,
IdempotencyKey: opts.IdempotencyKey,
@@ -300,6 +308,7 @@ func (c distributorUploadClient) Status(ctx context.Context, runID string) (runS
type uploadErrorContext struct {
Endpoint string
PipelineID string
BundleID string
IdempotencyKey string
SourcePath string
@@ -313,10 +322,10 @@ func wrapUploadError(err error, ctx uploadErrorContext) error {
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),
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),
}
}
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)
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)
}
func redactToken(err error, token string) error {

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

@@ -30,6 +30,7 @@ func TestUploadUsesConfiguredClientAndSingleFile(t *testing.T) {
client := newClient(cfg, factory.newClient)
result, err := client.Upload(context.Background(), UploadRequest{
PipelineID: "weatherreporter.daily",
BundleID: "weatherreporter.home.daily.run",
IdempotencyKey: "weatherreporter.home.daily.run",
SourcePath: "/tmp/report.md",
@@ -55,6 +56,9 @@ func TestUploadUsesConfiguredClientAndSingleFile(t *testing.T) {
t.Fatalf("factory timeout = %s, want 15s", factory.timeout)
}
got := factory.client.opts
if got.PipelineID != "weatherreporter.daily" {
t.Fatalf("PipelineID = %q, want weatherreporter.daily", got.PipelineID)
}
if got.BundleID != "weatherreporter.home.daily.run" {
t.Fatalf("BundleID = %q, want weatherreporter.home.daily.run", got.BundleID)
}
@@ -91,6 +95,13 @@ func TestUploadRejectsMissingInputs(t *testing.T) {
},
wantErr: "token environment variable",
},
{
name: "PipelineID",
mutate: func(c *Client, req *UploadRequest) {
req.PipelineID = ""
},
wantErr: "pipeline id is required",
},
{
name: "SourcePath",
mutate: func(c *Client, req *UploadRequest) {
@@ -170,7 +181,7 @@ func TestUploadWrapsUploadFailureWithContextWithoutToken(t *testing.T) {
if err == nil {
t.Fatal("Upload() error = nil, want error")
}
for _, want := range []string{cfg.Endpoint, req.BundleID, req.IdempotencyKey, req.SourcePath, req.BundlePath} {
for _, want := range []string{cfg.Endpoint, req.PipelineID, req.BundleID, req.IdempotencyKey, req.SourcePath, req.BundlePath} {
if !strings.Contains(err.Error(), want) {
t.Fatalf("error = %q, want context %q", err.Error(), want)
}
@@ -318,6 +329,7 @@ func TestUploadPreservesIdempotencyConflictDiagnosis(t *testing.T) {
func validUploadRequest() UploadRequest {
return UploadRequest{
PipelineID: "weatherreporter.daily",
BundleID: "weatherreporter.home.daily.run",
IdempotencyKey: "weatherreporter.home.daily.run",
SourcePath: "/tmp/report.md",

52
internal/app/app.go
View File

@@ -114,24 +114,25 @@ type BatchResult struct {
}
type BatchReportResult struct {
ReportID report.ID `json:"reportId"`
ReportName string `json:"reportName"`
PromptID string `json:"promptId"`
RunID string `json:"runId"`
Status string `json:"status"`
Error string `json:"error,omitempty"`
NotificationStatus string `json:"notificationStatus,omitempty"`
NotificationRunID string `json:"notificationRunId,omitempty"`
NotificationError string `json:"notificationError,omitempty"`
NotificationPath string `json:"notificationPath,omitempty"`
GeneratedAt time.Time `json:"generatedAt"`
ValidPeriod timeutil.Period `json:"validPeriod"`
BriefingPath string `json:"briefingPath,omitempty"`
DataPackagePath string `json:"dataPackagePath,omitempty"`
PreflightPath string `json:"preflightPath,omitempty"`
ReportPath string `json:"reportPath,omitempty"`
OutputPath string `json:"outputPath,omitempty"`
MetadataPath string `json:"metadataPath,omitempty"`
ReportID report.ID `json:"reportId"`
ReportName string `json:"reportName"`
PromptID string `json:"promptId"`
RunID string `json:"runId"`
Status string `json:"status"`
Error string `json:"error,omitempty"`
NotificationStatus string `json:"notificationStatus,omitempty"`
NotificationRunID string `json:"notificationRunId,omitempty"`
NotificationPipelineID string `json:"notificationPipelineId,omitempty"`
NotificationError string `json:"notificationError,omitempty"`
NotificationPath string `json:"notificationPath,omitempty"`
GeneratedAt time.Time `json:"generatedAt"`
ValidPeriod timeutil.Period `json:"validPeriod"`
BriefingPath string `json:"briefingPath,omitempty"`
DataPackagePath string `json:"dataPackagePath,omitempty"`
PreflightPath string `json:"preflightPath,omitempty"`
ReportPath string `json:"reportPath,omitempty"`
OutputPath string `json:"outputPath,omitempty"`
MetadataPath string `json:"metadataPath,omitempty"`
}
type BatchError struct {
@@ -157,6 +158,7 @@ type Notifier interface {
type NotificationRequest struct {
ReportID report.ID
RunID string
PipelineID string
BundleID string
IdempotencyKey string
ReportPath string
@@ -280,6 +282,7 @@ func RunBatchDetailed(ctx context.Context, req BatchRequest) (*BatchResult, erro
if errors.As(err, &notificationErr) {
item.NotificationStatus = "failed"
item.NotificationError = notificationErr.Error()
item.NotificationPipelineID = notificationErr.Request.PipelineID
if paths, pathErr := store.Paths(resolved); pathErr == nil {
item.NotificationPath = paths.Notification
}
@@ -297,6 +300,7 @@ func RunBatchDetailed(ctx context.Context, req BatchRequest) (*BatchResult, erro
if reportResult.Notification != nil {
item.NotificationStatus = reportResult.Notification.Status
item.NotificationRunID = reportResult.Notification.RunID
item.NotificationPipelineID = reportResult.Notification.PipelineID
}
result.Succeeded++
}
@@ -638,6 +642,10 @@ func buildNotificationRequest(cfg config.Config, resolved report.Resolved, repor
return NotificationRequest{}, err
}
values.BundleID = bundleID
pipelineID, err := config.RenderDistributorPipelineID(cfg.Notify.Distributor.PipelineIDTemplate, values)
if err != nil {
return NotificationRequest{}, err
}
idempotencyKey, err := config.RenderDistributorIdempotencyKey(cfg.Notify.Distributor.IdempotencyKeyTemplate, values)
if err != nil {
return NotificationRequest{}, err
@@ -649,6 +657,7 @@ func buildNotificationRequest(cfg config.Config, resolved report.Resolved, repor
return NotificationRequest{
ReportID: resolved.Definition.ID,
RunID: metadata.RunID,
PipelineID: pipelineID,
BundleID: bundleID,
IdempotencyKey: idempotencyKey,
ReportPath: reportPath,
@@ -667,6 +676,7 @@ func saveNotificationArtifact(ctx context.Context, store state.Store, resolved r
ReportID: resolved.Definition.ID,
AttemptedAt: time.Now(),
Endpoint: cfg.Notify.Distributor.Endpoint,
PipelineID: req.PipelineID,
BundleID: req.BundleID,
IdempotencyKey: req.IdempotencyKey,
SourcePath: req.ReportPath,
@@ -716,6 +726,7 @@ type distributorNotifier struct {
func (n distributorNotifier) Notify(ctx context.Context, req NotificationRequest) (*NotificationResult, error) {
result, err := n.client.Upload(ctx, distributoradapter.UploadRequest{
PipelineID: req.PipelineID,
BundleID: req.BundleID,
IdempotencyKey: req.IdempotencyKey,
SourcePath: req.ReportPath,
@@ -723,6 +734,7 @@ func (n distributorNotifier) Notify(ctx context.Context, req NotificationRequest
CreatedAt: req.CreatedAt,
})
notification := &NotificationResult{
PipelineID: req.PipelineID,
BundleID: req.BundleID,
IdempotencyKey: req.IdempotencyKey,
RunID: result.RunID,
@@ -731,7 +743,9 @@ func (n distributorNotifier) Notify(ctx context.Context, req NotificationRequest
StatusError: result.StatusError,
}
if result.RunStatus != nil {
notification.PipelineID = result.RunStatus.PipelineID
if result.RunStatus.PipelineID != "" {
notification.PipelineID = result.RunStatus.PipelineID
}
notification.AcceptedAt = result.RunStatus.AcceptedAt
notification.StartedAt = result.RunStatus.StartedAt
notification.FinishedAt = result.RunStatus.FinishedAt

32
internal/app/app_test.go
View File

@@ -288,6 +288,7 @@ func TestGenerateReportNotifiesManagedReportPath(t *testing.T) {
cfg := dailyTestConfig(t, server)
cfg.Workspace.Root = t.TempDir()
cfg.Notify.Distributor.Enabled = true
cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}"
resolved, err := ResolveGenerate(GenerateRequest{
Config: cfg,
Report: ReportDaily,
@@ -330,8 +331,12 @@ func TestGenerateReportNotifiesManagedReportPath(t *testing.T) {
if err != nil {
t.Fatalf("read notification artifact: %v", err)
}
if !strings.Contains(string(notificationData), `"replace_older"`) || !strings.Contains(string(notificationData), `"bundleCreated"`) {
t.Fatalf("notification artifact missing status report or created timestamp:\n%s", string(notificationData))
var notificationArtifact state.DistributorNotificationArtifact
if err := json.Unmarshal(notificationData, &notificationArtifact); err != nil {
t.Fatalf("decode notification artifact: %v", err)
}
if notificationArtifact.PipelineID != "weatherreporter.daily" || notificationArtifact.BundleCreated.IsZero() || notificationArtifact.RunStatus == nil || !strings.Contains(string(notificationArtifact.RunStatus.Report), "replace_older") {
t.Fatalf("notification artifact = %#v, want requested pipeline, status report, and created timestamp", notificationArtifact)
}
if len(notifier.requests) != 1 {
t.Fatalf("notification requests = %d, want 1", len(notifier.requests))
@@ -346,11 +351,14 @@ func TestGenerateReportNotifiesManagedReportPath(t *testing.T) {
if req.BundlePath != "daily.md" {
t.Fatalf("notification BundlePath = %q, want daily.md", req.BundlePath)
}
if req.BundleID != "weatherreporter.home.daily_today."+result.Metadata.RunID {
if req.PipelineID != "weatherreporter.daily" {
t.Fatalf("notification PipelineID = %q, want rendered pipeline", req.PipelineID)
}
if req.BundleID != "weatherreporter.home.daily_today" {
t.Fatalf("notification BundleID = %q, want default template", req.BundleID)
}
if req.IdempotencyKey != req.BundleID {
t.Fatalf("IdempotencyKey = %q, want bundle id %q", req.IdempotencyKey, req.BundleID)
if req.IdempotencyKey != req.BundleID+"."+result.Metadata.RunID {
t.Fatalf("IdempotencyKey = %q, want per-run key", req.IdempotencyKey)
}
if req.RunID != result.Metadata.RunID {
t.Fatalf("notification RunID = %q, want report run id %q", req.RunID, result.Metadata.RunID)
@@ -365,6 +373,7 @@ func TestGenerateReportNotificationFailureFailsReport(t *testing.T) {
cfg := dailyTestConfig(t, server)
cfg.Workspace.Root = t.TempDir()
cfg.Notify.Distributor.Enabled = true
cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}"
resolved, err := ResolveGenerate(GenerateRequest{
Config: cfg,
Report: ReportDaily,
@@ -441,6 +450,7 @@ func TestGenerateReportDoesNotNotifyAfterRenderOrRunFailure(t *testing.T) {
cfg := dailyTestConfig(t, server)
cfg.Workspace.Root = t.TempDir()
cfg.Notify.Distributor.Enabled = true
cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}"
resolved, err := ResolveGenerate(GenerateRequest{
Config: cfg,
Report: ReportDaily,
@@ -473,6 +483,7 @@ func TestGenerateReportDoesNotNotifyAfterFetchFailure(t *testing.T) {
cfg.WeatherAPI.Timezone = "America/Chicago"
cfg.Workspace.Root = t.TempDir()
cfg.Notify.Distributor.Enabled = true
cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}"
resolved, err := ResolveGenerate(GenerateRequest{
Config: cfg,
Report: ReportDaily,
@@ -1257,6 +1268,7 @@ func TestRunBatchContinuesAfterNotificationFailure(t *testing.T) {
cfg.WeatherAPI.Timezone = "America/Chicago"
cfg.Workspace.Root = t.TempDir()
cfg.Notify.Distributor.Enabled = true
cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}"
notifier := &recordingNotifier{
errByReport: map[report.ID]error{
report.ThreeDay: errors.New("distributor unavailable"),
@@ -1289,6 +1301,9 @@ func TestRunBatchContinuesAfterNotificationFailure(t *testing.T) {
if item.NotificationStatus != "failed" {
t.Fatalf("3-day notification status = %q, want failed", item.NotificationStatus)
}
if item.NotificationPipelineID != "weatherreporter.three-day" {
t.Fatalf("3-day notification pipeline = %q, want weatherreporter.three-day", item.NotificationPipelineID)
}
if !strings.Contains(item.NotificationError, "distributor unavailable") {
t.Fatalf("3-day notification error = %q, want distributor unavailable", item.NotificationError)
}
@@ -1300,6 +1315,9 @@ func TestRunBatchContinuesAfterNotificationFailure(t *testing.T) {
if item.NotificationStatus != "accepted" {
t.Fatalf("report %s notification status = %q, want accepted", item.ReportID, item.NotificationStatus)
}
if item.NotificationPipelineID == "" {
t.Fatalf("report %s notification pipeline is empty", item.ReportID)
}
}
if !failedThreeDay {
t.Fatalf("reports = %#v, want notification failure on 3-day item", result.Reports)
@@ -1500,9 +1518,13 @@ func (n *recordingNotifier) Notify(_ context.Context, req NotificationRequest) (
if result.IdempotencyKey == "" {
result.IdempotencyKey = req.IdempotencyKey
}
if result.PipelineID == "" {
result.PipelineID = req.PipelineID
}
return &result, nil
}
return &NotificationResult{
PipelineID: req.PipelineID,
BundleID: req.BundleID,
IdempotencyKey: req.IdempotencyKey,
Status: "accepted",

15
internal/cli/root_test.go
View File

@@ -469,7 +469,12 @@ func TestRunEveningUsesOutputDirectoryAndSummary(t *testing.T) {
func TestRunEveningReportsNotificationSuccess(t *testing.T) {
server := dailyServer(t)
distributorServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/upload" {
if r.URL.Path == "/runs/distributor-run-1" {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"run_id":"distributor-run-1","pipeline_id":"weatherreporter.daily","status":"succeeded","report":{"actions":[{"action":"replace_older"}]}}`))
return
}
if r.URL.Path != "/v1/pipelines/weatherreporter.daily/upload" {
http.NotFound(w, r)
return
}
@@ -481,7 +486,7 @@ func TestRunEveningReportsNotificationSuccess(t *testing.T) {
scriptoriumPath := writeFakeScriptorium(t, tempDir)
configPath := filepath.Join(tempDir, "config.yml")
workspaceRoot := filepath.Join(tempDir, "workspace")
configBody := "weather_api:\n base_url: " + server.URL + "/\n timezone: America/Chicago\nscriptorium:\n binary: " + scriptoriumPath + "\nworkspace:\n root: " + workspaceRoot + "\nnotify:\n distributor:\n enabled: true\n endpoint: " + distributorServer.URL + "\n token_env: CLI_DISTRIBUTOR_TOKEN\n"
configBody := "weather_api:\n base_url: " + server.URL + "/\n timezone: America/Chicago\nscriptorium:\n binary: " + scriptoriumPath + "\nworkspace:\n root: " + workspaceRoot + "\nnotify:\n distributor:\n enabled: true\n endpoint: " + distributorServer.URL + "\n token_env: CLI_DISTRIBUTOR_TOKEN\n pipeline_id_template: weatherreporter.{artifact_group}\n"
if err := os.WriteFile(configPath, []byte(configBody), 0o600); err != nil {
t.Fatalf("write config: %v", err)
}
@@ -505,10 +510,10 @@ func TestRunEveningReportsNotificationSuccess(t *testing.T) {
if len(summary.Reports) != 1 {
t.Fatalf("reports = %#v, want one report", summary.Reports)
}
if summary.Reports[0].NotificationStatus != "accepted" || summary.Reports[0].NotificationRunID != "distributor-run-1" {
if summary.Reports[0].NotificationStatus != "succeeded" || summary.Reports[0].NotificationRunID != "distributor-run-1" || summary.Reports[0].NotificationPipelineID != "weatherreporter.daily" {
t.Fatalf("notification fields = %#v", summary.Reports[0])
}
if !strings.Contains(stderr.String(), `notificationStatus="accepted"`) || !strings.Contains(stderr.String(), `notificationRunId="distributor-run-1"`) {
if !strings.Contains(stderr.String(), `notificationStatus="succeeded"`) || !strings.Contains(stderr.String(), `notificationRunId="distributor-run-1"`) {
t.Fatalf("stderr missing notification fields:\n%s", stderr.String())
}
if strings.Contains(stdout.String(), "cli-secret-token") || strings.Contains(stderr.String(), "cli-secret-token") {
@@ -527,7 +532,7 @@ func TestRunEveningReportsNotificationFailureWithoutToken(t *testing.T) {
scriptoriumPath := writeFakeScriptorium(t, tempDir)
configPath := filepath.Join(tempDir, "config.yml")
workspaceRoot := filepath.Join(tempDir, "workspace")
configBody := "weather_api:\n base_url: " + server.URL + "/\n timezone: America/Chicago\nscriptorium:\n binary: " + scriptoriumPath + "\nworkspace:\n root: " + workspaceRoot + "\nnotify:\n distributor:\n enabled: true\n endpoint: " + distributorServer.URL + "\n token_env: CLI_DISTRIBUTOR_TOKEN\n"
configBody := "weather_api:\n base_url: " + server.URL + "/\n timezone: America/Chicago\nscriptorium:\n binary: " + scriptoriumPath + "\nworkspace:\n root: " + workspaceRoot + "\nnotify:\n distributor:\n enabled: true\n endpoint: " + distributorServer.URL + "\n token_env: CLI_DISTRIBUTOR_TOKEN\n pipeline_id_template: weatherreporter.{artifact_group}\n"
if err := os.WriteFile(configPath, []byte(configBody), 0o600); err != nil {
t.Fatalf("write config: %v", err)
}

1
internal/config/config.go
View File

@@ -56,6 +56,7 @@ type DistributorNotifyConfig struct {
TokenEnv string `yaml:"token_env"`
Timeout time.Duration `yaml:"timeout"`
FailurePolicy NotifyFailurePolicy `yaml:"failure_policy"`
PipelineIDTemplate string `yaml:"pipeline_id_template"`
BundleIDTemplate string `yaml:"bundle_id_template"`
IdempotencyKeyTemplate string `yaml:"idempotency_key_template"`
ReportPathTemplate string `yaml:"report_path_template"`

53
internal/config/config_test.go
View File

@@ -44,10 +44,13 @@ func TestDefaults(t *testing.T) {
if cfg.Notify.Distributor.FailurePolicy != NotifyFailureError {
t.Fatalf("Notify.Distributor.FailurePolicy = %q, want error", cfg.Notify.Distributor.FailurePolicy)
}
if cfg.Notify.Distributor.BundleIDTemplate != "weatherreporter.{location_id}.{report_id}.{run_id}" {
if cfg.Notify.Distributor.PipelineIDTemplate != "" {
t.Fatalf("Notify.Distributor.PipelineIDTemplate = %q, want empty", cfg.Notify.Distributor.PipelineIDTemplate)
}
if cfg.Notify.Distributor.BundleIDTemplate != "weatherreporter.{location_id}.{report_id}" {
t.Fatalf("Notify.Distributor.BundleIDTemplate = %q, want default", cfg.Notify.Distributor.BundleIDTemplate)
}
if cfg.Notify.Distributor.IdempotencyKeyTemplate != "{bundle_id}" {
if cfg.Notify.Distributor.IdempotencyKeyTemplate != "{bundle_id}.{run_id}" {
t.Fatalf("Notify.Distributor.IdempotencyKeyTemplate = %q, want default", cfg.Notify.Distributor.IdempotencyKeyTemplate)
}
if cfg.Notify.Distributor.ReportPathTemplate != "{batch_output_name}" {
@@ -76,6 +79,9 @@ func TestLoadExampleConfig(t *testing.T) {
if cfg.Location.ID != "home" || cfg.Location.Name != "Brentwood" || cfg.Location.Region != "St. Louis Metro" {
t.Fatalf("Location = %#v, want example location", cfg.Location)
}
if cfg.Notify.Distributor.PipelineIDTemplate != "weatherreporter.{artifact_group}" {
t.Fatalf("PipelineIDTemplate = %q, want example pipeline template", cfg.Notify.Distributor.PipelineIDTemplate)
}
}
func TestLoadMinimalExampleConfig(t *testing.T) {
@@ -200,6 +206,27 @@ func TestEnabledDistributorNotifyValidation(t *testing.T) {
},
wantErr: "notify.distributor.failure_policy",
},
{
name: "PipelineTemplateEmpty",
mutate: func(cfg *Config) {
cfg.Notify.Distributor.PipelineIDTemplate = ""
},
wantErr: "notify.distributor.pipeline_id_template",
},
{
name: "PipelineTemplateUnknown",
mutate: func(cfg *Config) {
cfg.Notify.Distributor.PipelineIDTemplate = "{unknown}"
},
wantErr: "notify.distributor.pipeline_id_template",
},
{
name: "PipelineTemplateRenderedEmpty",
mutate: func(cfg *Config) {
cfg.Notify.Distributor.PipelineIDTemplate = " "
},
wantErr: "notify.distributor.pipeline_id_template",
},
{
name: "BundleTemplate",
mutate: func(cfg *Config) {
@@ -234,6 +261,7 @@ func TestEnabledDistributorNotifyValidation(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
cfg := Defaults()
cfg.Notify.Distributor.Enabled = true
cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}"
tt.mutate(&cfg)
err := Validate(cfg)
@@ -254,23 +282,31 @@ func TestDistributorTemplateRendering(t *testing.T) {
RunID: "20260607T120000Z",
ArtifactGroup: "daily",
BatchOutputName: "daily.md",
BundleID: "weatherreporter.home.daily.20260607T120000Z",
BundleID: "weatherreporter.home.daily",
}
bundleID, err := RenderDistributorBundleID("weatherreporter.{location_id}.{report_id}.{run_id}", values)
bundleID, err := RenderDistributorBundleID("weatherreporter.{location_id}.{report_id}", values)
if err != nil {
t.Fatalf("RenderDistributorBundleID() error = %v", err)
}
if bundleID != "weatherreporter.home.daily.20260607T120000Z" {
if bundleID != "weatherreporter.home.daily" {
t.Fatalf("bundleID = %q, want rendered value", bundleID)
}
idempotencyKey, err := RenderDistributorIdempotencyKey("{bundle_id}", values)
pipelineID, err := RenderDistributorPipelineID("weatherreporter.{artifact_group}.{bundle_id}", values)
if err != nil {
t.Fatalf("RenderDistributorPipelineID() error = %v", err)
}
if pipelineID != "weatherreporter.daily.weatherreporter.home.daily" {
t.Fatalf("pipelineID = %q, want rendered pipeline ID", pipelineID)
}
idempotencyKey, err := RenderDistributorIdempotencyKey("{bundle_id}.{run_id}", values)
if err != nil {
t.Fatalf("RenderDistributorIdempotencyKey() error = %v", err)
}
if idempotencyKey != "weatherreporter.home.daily.20260607T120000Z" {
t.Fatalf("idempotencyKey = %q, want rendered bundle ID", idempotencyKey)
t.Fatalf("idempotencyKey = %q, want rendered run key", idempotencyKey)
}
reportPath, err := RenderDistributorReportPath("reports/{batch_output_name}", values)
@@ -375,7 +411,8 @@ func TestLoadFileLoadsSecretsBeforeReturningNotifyConfig(t *testing.T) {
" directory: " + secretsDir + "\n" +
"notify:\n" +
" distributor:\n" +
" enabled: true\n"
" enabled: true\n" +
" pipeline_id_template: weatherreporter.{artifact_group}\n"
if err := os.WriteFile(path, []byte(configYAML), 0o600); err != nil {
t.Fatalf("write config fixture: %v", err)
}

5
internal/config/defaults.go
View File

@@ -28,8 +28,9 @@ func Defaults() Config {
TokenEnv: "DISTRIBUTOR_UPLOAD_TOKEN",
Timeout: 30 * time.Second,
FailurePolicy: NotifyFailureError,
BundleIDTemplate: "weatherreporter.{location_id}.{report_id}.{run_id}",
IdempotencyKeyTemplate: "{bundle_id}",
PipelineIDTemplate: "",
BundleIDTemplate: "weatherreporter.{location_id}.{report_id}",
IdempotencyKeyTemplate: "{bundle_id}.{run_id}",
ReportPathTemplate: "{batch_output_name}",
},
},

13
internal/config/notify_templates.go
View File

@@ -32,10 +32,23 @@ var distributorIdempotencyTemplateVariables = map[string]struct{}{
"bundle_id": {},
}
var distributorPipelineTemplateVariables = distributorIdempotencyTemplateVariables
func RenderDistributorBundleID(template string, values DistributorTemplateValues) (string, error) {
return renderDistributorTemplate("notify.distributor.bundle_id_template", template, values, distributorTemplateVariables)
}
func RenderDistributorPipelineID(template string, values DistributorTemplateValues) (string, error) {
rendered, err := renderDistributorTemplate("notify.distributor.pipeline_id_template", template, values, distributorPipelineTemplateVariables)
if err != nil {
return "", err
}
if strings.TrimSpace(rendered) == "" {
return "", fmt.Errorf("notify.distributor.pipeline_id_template renders an empty pipeline id")
}
return rendered, nil
}
func RenderDistributorIdempotencyKey(template string, values DistributorTemplateValues) (string, error) {
return renderDistributorTemplate("notify.distributor.idempotency_key_template", template, values, distributorIdempotencyTemplateVariables)
}

14
internal/config/validate.go
View File

@@ -100,6 +100,12 @@ func validateDistributorNotify(cfg DistributorNotifyConfig) error {
if cfg.FailurePolicy != NotifyFailureError {
return fmt.Errorf("notify.distributor.failure_policy must be error when enabled")
}
if cfg.PipelineIDTemplate == "" {
return fmt.Errorf("notify.distributor.pipeline_id_template is required when enabled")
}
if err := validateDistributorTemplate("notify.distributor.pipeline_id_template", cfg.PipelineIDTemplate, distributorPipelineTemplateVariables); err != nil {
return err
}
if cfg.BundleIDTemplate == "" {
return fmt.Errorf("notify.distributor.bundle_id_template is required when enabled")
}
@@ -122,6 +128,14 @@ func validateDistributorNotify(cfg DistributorNotifyConfig) error {
ArtifactGroup: "artifact",
BatchOutputName: "report.md",
}
bundleID, err := RenderDistributorBundleID(cfg.BundleIDTemplate, values)
if err != nil {
return err
}
values.BundleID = bundleID
if _, err := RenderDistributorPipelineID(cfg.PipelineIDTemplate, values); err != nil {
return err
}
if _, err := RenderDistributorReportPath(cfg.ReportPathTemplate, values); err != nil {
return err
}

3
internal/state/filesystem_test.go
View File

@@ -65,6 +65,7 @@ func TestSaveArtifactsAndMetadataRoundTrip(t *testing.T) {
ReportID: resolved.Definition.ID,
AttemptedAt: resolved.GeneratedAt,
Endpoint: "https://distributor.example.test",
PipelineID: "weatherreporter.daily",
BundleID: "weatherreporter.home.daily.run",
IdempotencyKey: "weatherreporter.home.daily.run",
SourcePath: "/tmp/report.md",
@@ -102,7 +103,7 @@ func TestSaveArtifactsAndMetadataRoundTrip(t *testing.T) {
if err := json.Unmarshal(notificationData, &notification); err != nil {
t.Fatalf("decode notification: %v", err)
}
if notification.SchemaVersion != DistributorNotificationSchemaVersion || notification.RunStatus == nil || notification.RunStatus.Status != "succeeded" {
if notification.SchemaVersion != DistributorNotificationSchemaVersion || notification.PipelineID != "weatherreporter.daily" || notification.RunStatus == nil || notification.RunStatus.Status != "succeeded" {
t.Fatalf("notification = %#v, want persisted distributor status", notification)
}
paths, err := store.Paths(resolved)

1
internal/state/store.go
View File

@@ -45,6 +45,7 @@ type DistributorNotificationArtifact struct {
ReportID report.ID `json:"reportId"`
AttemptedAt time.Time `json:"attemptedAt"`
Endpoint string `json:"endpoint"`
PipelineID string `json:"pipelineId,omitempty"`
BundleID string `json:"bundleId,omitempty"`
IdempotencyKey string `json:"idempotencyKey,omitempty"`
SourcePath string `json:"sourcePath,omitempty"`