Add batch distributor notification config

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`.

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

internal/config/config.go

@@ -67,6 +67,14 @@ type DistributorNotifyConfig struct {
 	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 {

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

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{

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)

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: