diff --git a/internal/app/app.go b/internal/app/app.go index 29aa7ff..2e830a2 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -879,14 +879,8 @@ func reportNotifier(cfg config.Config, notifier Notifier) (Notifier, bool) { } func buildNotificationRequest(cfg config.Config, resolved report.Resolved, reportPath string, metadata state.Metadata) (NotificationRequest, error) { - values := config.DistributorTemplateValues{ - LocationID: cfg.Location.ID, - ReportID: string(resolved.Definition.ID), - RunID: metadata.RunID, - ArtifactGroup: resolved.Definition.ArtifactGroup, - BatchOutputName: resolved.Definition.BatchOutputName, - } - if err := addDistributorValidPeriodValues(&values, resolved.ValidPeriod, cfg.WeatherAPI.Timezone); err != nil { + values, err := distributorTemplateValuesForReport(cfg, resolved, metadata.RunID, resolved.Definition.BatchOutputName) + if err != nil { return NotificationRequest{}, err } bundleID, err := config.RenderDistributorBundleID(cfg.Notify.Distributor.BundleIDTemplate, values) @@ -918,6 +912,23 @@ func buildNotificationRequest(cfg config.Config, resolved report.Resolved, repor }, nil } +func distributorTemplateValuesForReport(cfg config.Config, resolved report.Resolved, runID string, batchOutputName string) (config.DistributorTemplateValues, error) { + values := config.DistributorTemplateValues{ + LocationID: cfg.Location.ID, + ReportID: string(resolved.Definition.ID), + RunID: runID, + ArtifactGroup: resolved.Definition.ArtifactGroup, + BatchOutputName: batchOutputName, + } + if values.BatchOutputName == "" { + values.BatchOutputName = resolved.Definition.BatchOutputName + } + if err := addDistributorValidPeriodValues(&values, resolved.ValidPeriod, cfg.WeatherAPI.Timezone); err != nil { + return config.DistributorTemplateValues{}, err + } + return values, nil +} + func addDistributorValidPeriodValues(values *config.DistributorTemplateValues, period timeutil.Period, timezone string) error { location, err := timeutil.LoadLocation(timezone) if err != nil { diff --git a/internal/app/app_test.go b/internal/app/app_test.go index 06d8596..669c1d1 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -2330,6 +2330,121 @@ func TestBatchResultJSONIncludesNotification(t *testing.T) { } } +func TestBuildBatchNotificationRequestIncludesEveningReports(t *testing.T) { + server := dailyBundleServer(t) + cfg := dailyNotificationConfig(t, server) + cfg.Notify.Distributor.ReportPathTemplates = []string{ + "archive/{valid_start_date}/{artifact_group}/{run_id}.md", + "latest/{batch_output_name}", + } + cfg.Notify.Distributor.Batch.PipelineIDTemplate = "weatherreporter.{batch}.{batch_started_date}" + cfg.Notify.Distributor.Batch.BundleIDTemplate = "weatherreporter.{location_id}.{batch}" + cfg.Notify.Distributor.Batch.IdempotencyKeyTemplate = "{bundle_id}.{batch_run_id}" + startedAt := mustParse("2026-05-29T18:00:00-05:00") + planned, reports := plannedBatchNotificationReports(t, cfg, BatchEvening, startedAt, "2026-05-31", "2026-06-01") + runID := batchRunID(startedAt, BatchEvening) + + req, err := buildBatchNotificationRequest(cfg, BatchEvening, runID, startedAt, reports, planned) + if err != nil { + t.Fatalf("buildBatchNotificationRequest() error = %v", err) + } + + if req.Batch != BatchEvening || req.RunID != runID { + t.Fatalf("batch identity = %s/%s, want %s/%s", req.Batch, req.RunID, BatchEvening, runID) + } + if req.PipelineID != "weatherreporter.evening.2026-05-29" { + t.Fatalf("PipelineID = %q, want rendered batch pipeline", req.PipelineID) + } + if req.BundleID != "weatherreporter.home.evening" { + t.Fatalf("BundleID = %q, want rendered batch bundle id", req.BundleID) + } + wantKey := "weatherreporter.home.evening." + runID + if req.IdempotencyKey != wantKey { + t.Fatalf("IdempotencyKey = %q, want %q", req.IdempotencyKey, wantKey) + } + if !req.CreatedAt.Equal(startedAt) { + t.Fatalf("CreatedAt = %s, want %s", req.CreatedAt, startedAt) + } + if len(req.IncludedReports) != 3 { + t.Fatalf("IncludedReports = %d, want 3", len(req.IncludedReports)) + } + if len(req.Files) != 6 { + t.Fatalf("Files = %d, want two mappings per report", len(req.Files)) + } + + wantBundlePaths := map[string]struct{}{} + for _, plannedReport := range planned { + resolved := plannedReport.Resolved + runID := resolved.Metadata().RunID + outputName := plannedReport.OutputCopyName + if outputName == "" { + outputName = resolved.Definition.BatchOutputName + } + wantBundlePaths[fmt.Sprintf("archive/%s/%s/%s.md", resolved.ValidPeriod.Start.In(mustLoadTestLocation(t, cfg.WeatherAPI.Timezone)).Format(timeutil.DateLayout), resolved.Definition.ArtifactGroup, runID)] = struct{}{} + wantBundlePaths["latest/"+outputName] = struct{}{} + } + gotBundlePaths := map[string]struct{}{} + gotSourcePaths := map[string]struct{}{} + for _, file := range req.Files { + gotBundlePaths[file.BundlePath] = struct{}{} + gotSourcePaths[file.SourcePath] = struct{}{} + } + for want := range wantBundlePaths { + if _, ok := gotBundlePaths[want]; !ok { + t.Fatalf("bundle paths = %#v, missing %q", gotBundlePaths, want) + } + } + for _, item := range reports { + if _, ok := gotSourcePaths[item.ReportPath]; !ok { + t.Fatalf("source paths = %#v, missing managed report path %q", gotSourcePaths, item.ReportPath) + } + } + + uploadReq := batchDistributorUploadRequest(req) + if uploadReq.PipelineID != req.PipelineID || uploadReq.BundleID != req.BundleID || uploadReq.IdempotencyKey != req.IdempotencyKey || !uploadReq.CreatedAt.Equal(startedAt) { + t.Fatalf("upload request = %#v, want batch notification identity", uploadReq) + } + if len(uploadReq.Files) != len(req.Files) { + t.Fatalf("upload files = %d, want %d", len(uploadReq.Files), len(req.Files)) + } +} + +func TestBuildBatchNotificationRequestRejectsDuplicateBundlePaths(t *testing.T) { + server := dailyBundleServer(t) + cfg := dailyNotificationConfig(t, server) + cfg.Notify.Distributor.ReportPathTemplates = []string{"index.md"} + startedAt := mustParse("2026-05-29T18:00:00-05:00") + planned, reports := plannedBatchNotificationReports(t, cfg, BatchEvening, startedAt, "2026-05-31") + + _, err := buildBatchNotificationRequest(cfg, BatchEvening, batchRunID(startedAt, BatchEvening), startedAt, reports, planned) + if err == nil { + t.Fatal("buildBatchNotificationRequest() error = nil, want duplicate path error") + } + for _, want := range []string{"duplicate bundle path", "index.md", "report", "run", "source path"} { + if !strings.Contains(err.Error(), want) { + t.Fatalf("error = %q, want %q", err.Error(), want) + } + } +} + +func TestBuildBatchNotificationRequestRejectsMissingReportPath(t *testing.T) { + server := dailyBundleServer(t) + cfg := dailyNotificationConfig(t, server) + startedAt := mustParse("2026-05-29T18:00:00-05:00") + planned, reports := plannedBatchNotificationReports(t, cfg, BatchEvening, startedAt, "2026-05-31") + reports[0].ReportPath = "" + + _, err := buildBatchNotificationRequest(cfg, BatchEvening, batchRunID(startedAt, BatchEvening), startedAt, reports, planned) + if err == nil { + t.Fatal("buildBatchNotificationRequest() error = nil, want missing path error") + } + for _, want := range []string{"missing managed report path", string(reports[0].ReportID), reports[0].RunID} { + if !strings.Contains(err.Error(), want) { + t.Fatalf("error = %q, want %q", err.Error(), want) + } + } +} + func TestRunBatchContinuesAfterReportFailure(t *testing.T) { server := dailyBundleServer(t) cfg := dailyWorkspaceConfig(t, server) @@ -2712,6 +2827,27 @@ func dailyNotificationConfig(t *testing.T, server *httptest.Server) config.Confi return cfg } +func plannedBatchNotificationReports(t *testing.T, cfg config.Config, batch BatchKind, now time.Time, futureDailyDates ...string) ([]plannedBatchReport, []BatchReportResult) { + t.Helper() + collection := collectionWithFutureDailyForTest(t, cfg, futureDailyDates...) + planned, err := planBatchRun(BatchRequest{Config: cfg, Batch: batch}, now, collection) + if err != nil { + t.Fatalf("planBatchRun() error = %v", err) + } + reportDir := filepath.Join(t.TempDir(), "managed-reports") + results := make([]BatchReportResult, 0, len(planned)) + for _, item := range planned { + metadata := item.Resolved.Metadata() + results = append(results, BatchReportResult{ + ReportID: item.Resolved.Definition.ID, + RunID: metadata.RunID, + Status: "succeeded", + ReportPath: filepath.Join(reportDir, string(item.Resolved.Definition.ID), metadata.RunID+".md"), + }) + } + return planned, results +} + func collectionForTest(t *testing.T, cfg config.Config) collect.Result { t.Helper() bundle, err := FetchBundle(context.Background(), FetchBundleRequest{Config: cfg}) diff --git a/internal/app/batch_notification.go b/internal/app/batch_notification.go index 59f9601..1815d82 100644 --- a/internal/app/batch_notification.go +++ b/internal/app/batch_notification.go @@ -4,7 +4,9 @@ import ( "fmt" "time" + distributoradapter "gitea.maximumdirect.net/eric/weatherreporter/internal/adapters/distributor" "gitea.maximumdirect.net/eric/weatherreporter/internal/config" + "gitea.maximumdirect.net/eric/weatherreporter/internal/report" "gitea.maximumdirect.net/eric/weatherreporter/internal/timeutil" ) @@ -16,10 +18,141 @@ type batchNotificationIdentity struct { IdempotencyKey string } +type batchNotificationRequest struct { + Batch BatchKind + RunID string + PipelineID string + BundleID string + IdempotencyKey string + Files []batchNotificationFile + IncludedReports []BatchNotificationReport + CreatedAt time.Time +} + +type batchNotificationFile struct { + ReportID report.ID + RunID string + SourcePath string + BundlePath string +} + func batchRunID(startedAt time.Time, batch BatchKind) string { return startedAt.UTC().Format(runIDTimestampLayout) + "_" + string(batch) } +func buildBatchNotificationRequest(cfg config.Config, batch BatchKind, runID string, startedAt time.Time, reports []BatchReportResult, planned []plannedBatchReport) (batchNotificationRequest, error) { + if len(reports) == 0 { + return batchNotificationRequest{}, fmt.Errorf("batch notification requires at least one report") + } + + identity, err := renderBatchNotificationIdentity(cfg, batch, runID, startedAt) + if err != nil { + return batchNotificationRequest{}, err + } + if identity.PipelineID == "" { + return batchNotificationRequest{}, fmt.Errorf("batch notification pipeline id is required") + } + if identity.BundleID == "" { + return batchNotificationRequest{}, fmt.Errorf("batch notification bundle id is required") + } + if identity.IdempotencyKey == "" { + return batchNotificationRequest{}, fmt.Errorf("batch notification idempotency key is required for bundle %q", identity.BundleID) + } + + plannedByRunID, err := plannedReportsByRunID(planned) + if err != nil { + return batchNotificationRequest{}, err + } + + req := batchNotificationRequest{ + Batch: batch, + RunID: runID, + PipelineID: identity.PipelineID, + BundleID: identity.BundleID, + IdempotencyKey: identity.IdempotencyKey, + CreatedAt: startedAt, + } + seenBundlePaths := map[string]batchNotificationFile{} + for _, item := range reports { + plannedReport, ok := plannedByRunID[item.RunID] + if !ok { + return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q has no matching planned report", item.ReportID, item.RunID) + } + if item.ReportID != plannedReport.Resolved.Definition.ID { + return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q does not match planned report %q", item.ReportID, item.RunID, plannedReport.Resolved.Definition.ID) + } + if item.ReportPath == "" { + return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q is missing managed report path", item.ReportID, item.RunID) + } + + values, err := distributorTemplateValuesForReport(cfg, plannedReport.Resolved, item.RunID, plannedReport.OutputCopyName) + if err != nil { + return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q source path %q: %w", item.ReportID, item.RunID, item.ReportPath, err) + } + bundlePaths, err := config.RenderDistributorReportPaths(cfg.Notify.Distributor.ReportPathTemplates, values) + if err != nil { + return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q source path %q: %w", item.ReportID, item.RunID, item.ReportPath, err) + } + + included := BatchNotificationReport{ + ReportID: item.ReportID, + RunID: item.RunID, + SourcePath: item.ReportPath, + BundlePaths: append([]string(nil), bundlePaths...), + } + for _, bundlePath := range bundlePaths { + file := batchNotificationFile{ + ReportID: item.ReportID, + RunID: item.RunID, + SourcePath: item.ReportPath, + BundlePath: bundlePath, + } + if previous, ok := seenBundlePaths[bundlePath]; ok { + return batchNotificationRequest{}, fmt.Errorf("batch notification duplicate bundle path %q for report %q run %q source path %q; already used by report %q run %q source path %q", bundlePath, item.ReportID, item.RunID, item.ReportPath, previous.ReportID, previous.RunID, previous.SourcePath) + } + seenBundlePaths[bundlePath] = file + req.Files = append(req.Files, file) + } + req.IncludedReports = append(req.IncludedReports, included) + } + if len(req.Files) == 0 { + return batchNotificationRequest{}, fmt.Errorf("batch notification requires at least one file mapping") + } + return req, nil +} + +func plannedReportsByRunID(planned []plannedBatchReport) (map[string]plannedBatchReport, error) { + byRunID := make(map[string]plannedBatchReport, len(planned)) + for _, item := range planned { + runID := item.Resolved.Metadata().RunID + if runID == "" { + return nil, fmt.Errorf("planned report %q has empty run id", item.Resolved.Definition.ID) + } + if previous, ok := byRunID[runID]; ok { + return nil, fmt.Errorf("planned reports %q and %q share run id %q", previous.Resolved.Definition.ID, item.Resolved.Definition.ID, runID) + } + byRunID[runID] = item + } + return byRunID, nil +} + +func batchDistributorUploadRequest(req batchNotificationRequest) distributoradapter.UploadRequest { + files := make([]distributoradapter.UploadFile, 0, len(req.Files)) + for _, file := range req.Files { + files = append(files, distributoradapter.UploadFile{ + SourcePath: file.SourcePath, + BundlePath: file.BundlePath, + }) + } + return distributoradapter.UploadRequest{ + PipelineID: req.PipelineID, + BundleID: req.BundleID, + IdempotencyKey: req.IdempotencyKey, + Files: files, + CreatedAt: req.CreatedAt, + } +} + func renderBatchNotificationIdentity(cfg config.Config, batch BatchKind, runID string, startedAt time.Time) (batchNotificationIdentity, error) { values, err := batchNotificationTemplateValues(cfg, batch, runID, startedAt) if err != nil {