diff --git a/docs/config.md b/docs/config.md index 8ace2c6..a96fa62 100644 --- a/docs/config.md +++ b/docs/config.md @@ -104,6 +104,17 @@ report rendering succeeds and final metadata is saved. ```yaml - "{valid_start_date}/{artifact_group}/{valid_start_date}-{artifact_group}-{run_id}.md" ``` +- `batch.enabled`: whether batch distributor notification config is active + when distributor notification is enabled. Default: `true`. +- `batch.pipeline_id_template`: template for batch distributor pipeline IDs. + Required when distributor notification and batch notification are enabled. + Default: `weatherreporter`. +- `batch.bundle_id_template`: template for batch distributor bundle IDs. + Required when distributor notification and batch notification are enabled. + Default: `weatherreporter.{location_id}.{batch}`. +- `batch.idempotency_key_template`: template for batch distributor idempotency + keys. Required when distributor notification and batch notification are + enabled. Default: `{bundle_id}.{batch_run_id}`. Supported template variables are `location_id`, `report_id`, `run_id`, `artifact_group`, `batch_output_name`, `valid_start_date`, `valid_end_date`, @@ -116,6 +127,11 @@ The rendered pipeline ID selects the configured distributor `http_upload` workflow. The rendered bundle ID is the stable logical source identity for the report stream. The rendered idempotency key is the per-run retry identity. +Batch templates support `location_id`, `batch`, `batch_run_id`, and +`batch_started_date`. Batch idempotency templates may also use `bundle_id`. +Batch bundle IDs identify a logical batch stream; batch idempotency keys +identify a specific retryable batch attempt. + Rendered report paths must be unique relative paths with `/` separators. They must not contain backslashes, empty path segments, `.`, `..`, `manifest.json`, or `.distributor.json`. diff --git a/examples/config.yml b/examples/config.yml index 3f5bdae..c083411 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -26,6 +26,11 @@ notify: idempotency_key_template: "{bundle_id}.{run_id}" report_path_templates: - "{valid_start_date}/{artifact_group}/{valid_start_date}-{artifact_group}-{run_id}.md" + batch: + enabled: true + pipeline_id_template: "weatherreporter" + bundle_id_template: "weatherreporter.{location_id}.{batch}" + idempotency_key_template: "{bundle_id}.{batch_run_id}" missing_source: default: warn diff --git a/internal/config/config.go b/internal/config/config.go index 207d151..34d9ec1 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -58,15 +58,23 @@ type NotifyConfig struct { } type DistributorNotifyConfig struct { - Enabled bool `yaml:"enabled"` - Endpoint string `yaml:"endpoint"` - 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"` - ReportPathTemplates []string `yaml:"report_path_templates"` + Enabled bool `yaml:"enabled"` + Endpoint string `yaml:"endpoint"` + 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"` + ReportPathTemplates []string `yaml:"report_path_templates"` + Batch DistributorBatchNotifyConfig `yaml:"batch"` +} + +type DistributorBatchNotifyConfig struct { + Enabled bool `yaml:"enabled"` + PipelineIDTemplate string `yaml:"pipeline_id_template"` + BundleIDTemplate string `yaml:"bundle_id_template"` + IdempotencyKeyTemplate string `yaml:"idempotency_key_template"` } type MissingSourceConfig struct { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index b9ba12e..3d08a6b 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -63,6 +63,18 @@ func TestDefaults(t *testing.T) { if strings.Join(cfg.Notify.Distributor.ReportPathTemplates, "\n") != strings.Join(wantReportPaths, "\n") { t.Fatalf("Notify.Distributor.ReportPathTemplates = %#v, want %#v", cfg.Notify.Distributor.ReportPathTemplates, wantReportPaths) } + if !cfg.Notify.Distributor.Batch.Enabled { + t.Fatalf("Notify.Distributor.Batch.Enabled = false, want true") + } + if cfg.Notify.Distributor.Batch.PipelineIDTemplate != "weatherreporter" { + t.Fatalf("Notify.Distributor.Batch.PipelineIDTemplate = %q, want weatherreporter", cfg.Notify.Distributor.Batch.PipelineIDTemplate) + } + if cfg.Notify.Distributor.Batch.BundleIDTemplate != "weatherreporter.{location_id}.{batch}" { + t.Fatalf("Notify.Distributor.Batch.BundleIDTemplate = %q, want default", cfg.Notify.Distributor.Batch.BundleIDTemplate) + } + if cfg.Notify.Distributor.Batch.IdempotencyKeyTemplate != "{bundle_id}.{batch_run_id}" { + t.Fatalf("Notify.Distributor.Batch.IdempotencyKeyTemplate = %q, want default", cfg.Notify.Distributor.Batch.IdempotencyKeyTemplate) + } if cfg.MissingSource.Default != MissingSourceWarn { t.Fatalf("MissingSource.Default = %q, want warn", cfg.MissingSource.Default) } @@ -92,6 +104,18 @@ func TestLoadExampleConfig(t *testing.T) { if len(cfg.Notify.Distributor.ReportPathTemplates) != 1 { t.Fatalf("ReportPathTemplates = %#v, want example archive path", cfg.Notify.Distributor.ReportPathTemplates) } + if !cfg.Notify.Distributor.Batch.Enabled { + t.Fatalf("Notify.Distributor.Batch.Enabled = false, want true") + } + if cfg.Notify.Distributor.Batch.PipelineIDTemplate != "weatherreporter" { + t.Fatalf("Batch PipelineIDTemplate = %q, want weatherreporter", cfg.Notify.Distributor.Batch.PipelineIDTemplate) + } + if cfg.Notify.Distributor.Batch.BundleIDTemplate != "weatherreporter.{location_id}.{batch}" { + t.Fatalf("Batch BundleIDTemplate = %q, want example batch bundle template", cfg.Notify.Distributor.Batch.BundleIDTemplate) + } + if cfg.Notify.Distributor.Batch.IdempotencyKeyTemplate != "{bundle_id}.{batch_run_id}" { + t.Fatalf("Batch IdempotencyKeyTemplate = %q, want example batch idempotency template", cfg.Notify.Distributor.Batch.IdempotencyKeyTemplate) + } overrides, err := cfg.ReportModuleOverrides() if err != nil { t.Fatalf("ReportModuleOverrides() error = %v", err) @@ -782,6 +806,18 @@ func TestDisabledDistributorNotifyAcceptsOmittedFields(t *testing.T) { } } +func TestDisabledDistributorNotifyAcceptsMalformedBatchTemplates(t *testing.T) { + cfg := Defaults() + cfg.Notify.Distributor.Enabled = false + cfg.Notify.Distributor.Batch.PipelineIDTemplate = "{unknown}" + cfg.Notify.Distributor.Batch.BundleIDTemplate = "{unknown}" + cfg.Notify.Distributor.Batch.IdempotencyKeyTemplate = "{unknown}" + + if err := Validate(cfg); err != nil { + t.Fatalf("Validate() error = %v", err) + } +} + func TestEnabledDistributorNotifyValidation(t *testing.T) { tests := []struct { name string @@ -906,6 +942,109 @@ func TestEnabledDistributorNotifyValidation(t *testing.T) { } } +func TestEnabledDistributorBatchNotifyValidation(t *testing.T) { + tests := []struct { + name string + mutate func(*Config) + wantErr string + }{ + { + name: "PipelineTemplateEmpty", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Batch.PipelineIDTemplate = "" + }, + wantErr: "notify.distributor.batch.pipeline_id_template", + }, + { + name: "PipelineTemplateUnknown", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Batch.PipelineIDTemplate = "{report_id}" + }, + wantErr: "notify.distributor.batch.pipeline_id_template", + }, + { + name: "PipelineTemplateRenderedEmpty", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Batch.PipelineIDTemplate = " " + }, + wantErr: "notify.distributor.batch.pipeline_id_template", + }, + { + name: "BundleTemplateEmpty", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Batch.BundleIDTemplate = "" + }, + wantErr: "notify.distributor.batch.bundle_id_template", + }, + { + name: "BundleTemplateUnknown", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Batch.BundleIDTemplate = "{run_id}" + }, + wantErr: "notify.distributor.batch.bundle_id_template", + }, + { + name: "BundleTemplateRenderedEmpty", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Batch.BundleIDTemplate = " " + }, + wantErr: "notify.distributor.batch.bundle_id_template", + }, + { + name: "IdempotencyTemplateEmpty", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Batch.IdempotencyKeyTemplate = "" + }, + wantErr: "notify.distributor.batch.idempotency_key_template", + }, + { + name: "IdempotencyTemplateUnknown", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Batch.IdempotencyKeyTemplate = "{report_id}" + }, + wantErr: "notify.distributor.batch.idempotency_key_template", + }, + { + name: "IdempotencyTemplateRenderedEmpty", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Batch.IdempotencyKeyTemplate = " " + }, + wantErr: "notify.distributor.batch.idempotency_key_template", + }, + } + + for _, tt := range tests { + 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) + if err == nil { + t.Fatal("Validate() error = nil, want error") + } + if !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("error = %q, want %q", err.Error(), tt.wantErr) + } + }) + } +} + +func TestDisabledDistributorBatchNotifySkipsBatchTemplateValidation(t *testing.T) { + cfg := Defaults() + cfg.Notify.Distributor.Enabled = true + cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}" + cfg.Notify.Distributor.Batch.Enabled = false + cfg.Notify.Distributor.Batch.PipelineIDTemplate = "{unknown}" + cfg.Notify.Distributor.Batch.BundleIDTemplate = "{unknown}" + cfg.Notify.Distributor.Batch.IdempotencyKeyTemplate = "{unknown}" + + if err := Validate(cfg); err != nil { + t.Fatalf("Validate() error = %v", err) + } +} + func TestDistributorTemplateRendering(t *testing.T) { values := DistributorTemplateValues{ LocationID: "home", @@ -962,6 +1101,69 @@ func TestDistributorTemplateRendering(t *testing.T) { } } +func TestDistributorBatchTemplateRendering(t *testing.T) { + values := DistributorBatchTemplateValues{ + LocationID: "home", + Batch: "evening", + BatchRunID: "20260617T235037.642224552Z_evening", + BatchStartedDate: "2026-06-17", + } + + bundleID, err := RenderDistributorBatchBundleID("weatherreporter.{location_id}.{batch}", values) + if err != nil { + t.Fatalf("RenderDistributorBatchBundleID() error = %v", err) + } + if bundleID != "weatherreporter.home.evening" { + t.Fatalf("bundleID = %q, want batch bundle ID", bundleID) + } + values.BundleID = bundleID + + pipelineID, err := RenderDistributorBatchPipelineID("weatherreporter", values) + if err != nil { + t.Fatalf("RenderDistributorBatchPipelineID() error = %v", err) + } + if pipelineID != "weatherreporter" { + t.Fatalf("pipelineID = %q, want weatherreporter", pipelineID) + } + + idempotencyKey, err := RenderDistributorBatchIdempotencyKey("{bundle_id}.{batch_run_id}", values) + if err != nil { + t.Fatalf("RenderDistributorBatchIdempotencyKey() error = %v", err) + } + if idempotencyKey != "weatherreporter.home.evening.20260617T235037.642224552Z_evening" { + t.Fatalf("idempotencyKey = %q, want batch retry key", idempotencyKey) + } + + bundleID, err = RenderDistributorBatchBundleID("weatherreporter.{batch_started_date}.{batch}", values) + if err != nil { + t.Fatalf("RenderDistributorBatchBundleID() with date error = %v", err) + } + if bundleID != "weatherreporter.2026-06-17.evening" { + t.Fatalf("bundleID = %q, want date-aware batch bundle ID", bundleID) + } +} + +func TestDistributorBatchTemplateRejectsUnknownAndMalformedVariables(t *testing.T) { + tests := []struct { + name string + template string + }{ + {name: "Unknown", template: "{report_id}"}, + {name: "Unclosed", template: "{batch"}, + {name: "Unopened", template: "batch}"}, + {name: "Empty", template: "{}"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := RenderDistributorBatchBundleID(tt.template, DistributorBatchTemplateValues{}) + if err == nil { + t.Fatal("RenderDistributorBatchBundleID() error = nil, want error") + } + }) + } +} + func TestDistributorTemplateRejectsUnknownAndMalformedVariables(t *testing.T) { tests := []struct { name string diff --git a/internal/config/defaults.go b/internal/config/defaults.go index 3bd74ee..418b3fc 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -34,6 +34,12 @@ func Defaults() Config { ReportPathTemplates: []string{ "{valid_start_date}/{artifact_group}/{valid_start_date}-{artifact_group}-{run_id}.md", }, + Batch: DistributorBatchNotifyConfig{ + Enabled: true, + PipelineIDTemplate: "weatherreporter", + BundleIDTemplate: "weatherreporter.{location_id}.{batch}", + IdempotencyKeyTemplate: "{bundle_id}.{batch_run_id}", + }, }, }, MissingSource: MissingSourceConfig{ diff --git a/internal/config/notify_templates.go b/internal/config/notify_templates.go index 3ce3ffd..759234f 100644 --- a/internal/config/notify_templates.go +++ b/internal/config/notify_templates.go @@ -21,6 +21,14 @@ type DistributorTemplateValues struct { BundleID string } +type DistributorBatchTemplateValues struct { + LocationID string + Batch string + BatchRunID string + BatchStartedDate string + BundleID string +} + var distributorTemplateVariables = map[string]struct{}{ "location_id": {}, "report_id": {}, @@ -52,6 +60,23 @@ var distributorIdempotencyTemplateVariables = map[string]struct{}{ var distributorPipelineTemplateVariables = distributorIdempotencyTemplateVariables +var distributorBatchTemplateVariables = map[string]struct{}{ + "location_id": {}, + "batch": {}, + "batch_run_id": {}, + "batch_started_date": {}, +} + +var distributorBatchIdempotencyTemplateVariables = map[string]struct{}{ + "location_id": {}, + "batch": {}, + "batch_run_id": {}, + "batch_started_date": {}, + "bundle_id": {}, +} + +var distributorBatchPipelineTemplateVariables = distributorBatchTemplateVariables + func RenderDistributorBundleID(template string, values DistributorTemplateValues) (string, error) { return renderDistributorTemplate("notify.distributor.bundle_id_template", template, values, distributorTemplateVariables) } @@ -71,6 +96,39 @@ func RenderDistributorIdempotencyKey(template string, values DistributorTemplate return renderDistributorTemplate("notify.distributor.idempotency_key_template", template, values, distributorIdempotencyTemplateVariables) } +func RenderDistributorBatchBundleID(template string, values DistributorBatchTemplateValues) (string, error) { + rendered, err := renderDistributorBatchTemplate("notify.distributor.batch.bundle_id_template", template, values, distributorBatchTemplateVariables) + if err != nil { + return "", err + } + if strings.TrimSpace(rendered) == "" { + return "", fmt.Errorf("notify.distributor.batch.bundle_id_template renders an empty bundle id") + } + return rendered, nil +} + +func RenderDistributorBatchPipelineID(template string, values DistributorBatchTemplateValues) (string, error) { + rendered, err := renderDistributorBatchTemplate("notify.distributor.batch.pipeline_id_template", template, values, distributorBatchPipelineTemplateVariables) + if err != nil { + return "", err + } + if strings.TrimSpace(rendered) == "" { + return "", fmt.Errorf("notify.distributor.batch.pipeline_id_template renders an empty pipeline id") + } + return rendered, nil +} + +func RenderDistributorBatchIdempotencyKey(template string, values DistributorBatchTemplateValues) (string, error) { + rendered, err := renderDistributorBatchTemplate("notify.distributor.batch.idempotency_key_template", template, values, distributorBatchIdempotencyTemplateVariables) + if err != nil { + return "", err + } + if strings.TrimSpace(rendered) == "" { + return "", fmt.Errorf("notify.distributor.batch.idempotency_key_template renders an empty idempotency key") + } + return rendered, nil +} + func RenderDistributorReportPaths(templates []string, values DistributorTemplateValues) ([]string, error) { if len(templates) == 0 { return nil, fmt.Errorf("notify.distributor.report_path_templates must contain at least one entry") @@ -100,6 +158,11 @@ func validateDistributorTemplate(name, template string, allowed map[string]struc return err } +func validateDistributorBatchTemplate(name, template string, allowed map[string]struct{}) error { + _, err := renderDistributorBatchTemplate(name, template, DistributorBatchTemplateValues{}, allowed) + return err +} + func renderDistributorTemplate(name, template string, values DistributorTemplateValues, allowed map[string]struct{}) (string, error) { var rendered strings.Builder for i := 0; i < len(template); { @@ -128,6 +191,34 @@ func renderDistributorTemplate(name, template string, values DistributorTemplate return rendered.String(), nil } +func renderDistributorBatchTemplate(name, template string, values DistributorBatchTemplateValues, allowed map[string]struct{}) (string, error) { + var rendered strings.Builder + for i := 0; i < len(template); { + switch template[i] { + case '{': + end := strings.IndexByte(template[i+1:], '}') + if end < 0 { + return "", fmt.Errorf("%s contains an unclosed template variable", name) + } + variable := template[i+1 : i+1+end] + if variable == "" { + return "", fmt.Errorf("%s contains an empty template variable", name) + } + if _, ok := allowed[variable]; !ok { + return "", fmt.Errorf("%s contains unknown template variable %q", name, variable) + } + rendered.WriteString(distributorBatchTemplateValue(variable, values)) + i += end + 2 + case '}': + return "", fmt.Errorf("%s contains an unopened template variable", name) + default: + rendered.WriteByte(template[i]) + i++ + } + } + return rendered.String(), nil +} + func distributorTemplateValue(variable string, values DistributorTemplateValues) string { switch variable { case "location_id": @@ -159,6 +250,23 @@ func distributorTemplateValue(variable string, values DistributorTemplateValues) } } +func distributorBatchTemplateValue(variable string, values DistributorBatchTemplateValues) string { + switch variable { + case "location_id": + return values.LocationID + case "batch": + return values.Batch + case "batch_run_id": + return values.BatchRunID + case "batch_started_date": + return values.BatchStartedDate + case "bundle_id": + return values.BundleID + default: + return "" + } +} + func ValidateDistributorReportPath(name, path string) error { if path == "" { return fmt.Errorf("%s renders an empty path", name) diff --git a/internal/config/validate.go b/internal/config/validate.go index 37d5bbd..a38b516 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -148,10 +148,56 @@ func validateDistributorNotify(cfg DistributorNotifyConfig) error { if _, err := RenderDistributorReportPaths(cfg.ReportPathTemplates, values); err != nil { return err } + if err := validateDistributorBatchNotify(cfg.Batch); err != nil { + return err + } return nil } +func validateDistributorBatchNotify(cfg DistributorBatchNotifyConfig) error { + if !cfg.Enabled { + return nil + } + if cfg.PipelineIDTemplate == "" { + return fmt.Errorf("notify.distributor.batch.pipeline_id_template is required when enabled") + } + if err := validateDistributorBatchTemplate("notify.distributor.batch.pipeline_id_template", cfg.PipelineIDTemplate, distributorBatchPipelineTemplateVariables); err != nil { + return err + } + if cfg.BundleIDTemplate == "" { + return fmt.Errorf("notify.distributor.batch.bundle_id_template is required when enabled") + } + if err := validateDistributorBatchTemplate("notify.distributor.batch.bundle_id_template", cfg.BundleIDTemplate, distributorBatchTemplateVariables); err != nil { + return err + } + if cfg.IdempotencyKeyTemplate == "" { + return fmt.Errorf("notify.distributor.batch.idempotency_key_template is required when enabled") + } + if err := validateDistributorBatchTemplate("notify.distributor.batch.idempotency_key_template", cfg.IdempotencyKeyTemplate, distributorBatchIdempotencyTemplateVariables); err != nil { + return err + } + + values := DistributorBatchTemplateValues{ + LocationID: "location", + Batch: "morning", + BatchRunID: "20260529T100000.000000000Z_morning", + BatchStartedDate: "2026-05-29", + } + bundleID, err := RenderDistributorBatchBundleID(cfg.BundleIDTemplate, values) + if err != nil { + return err + } + values.BundleID = bundleID + if _, err := RenderDistributorBatchPipelineID(cfg.PipelineIDTemplate, values); err != nil { + return err + } + if _, err := RenderDistributorBatchIdempotencyKey(cfg.IdempotencyKeyTemplate, values); err != nil { + return err + } + return nil +} + func validatePolicy(name string, policy MissingSourcePolicy) error { switch policy { case MissingSourceError, MissingSourceWarn, MissingSourceNone: