diff --git a/internal/app/app.go b/internal/app/app.go index 83285ef..65d3e60 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -176,9 +176,32 @@ func (e BatchError) Error() string { if e.Result == nil { return "batch failed" } + if batchNotificationFailed(e.Result) && batchReportFailures(e.Result) == 0 { + if e.Result.Notification.Error != "" { + return fmt.Sprintf("batch %s notification failed: %s", e.Result.Batch, e.Result.Notification.Error) + } + return fmt.Sprintf("batch %s notification failed", e.Result.Batch) + } return fmt.Sprintf("batch %s failed: %d of %d reports failed", e.Result.Batch, e.Result.Failed, e.Result.Total) } +func batchNotificationFailed(result *BatchResult) bool { + return result != nil && result.Notification != nil && result.Notification.Status == "failed" +} + +func batchReportFailures(result *BatchResult) int { + if result == nil { + return 0 + } + failures := 0 + for _, item := range result.Reports { + if item.Status == "failed" { + failures++ + } + } + return failures +} + type Renderer interface { Render(context.Context, scriptorium.RenderRequest) (*scriptorium.RenderResult, error) Run(context.Context, scriptorium.RunRequest) (*scriptorium.RunResult, error) @@ -365,6 +388,13 @@ func RunBatchDetailed(ctx context.Context, req BatchRequest) (*BatchResult, erro result.Reports = append(result.Reports, item) } result.Total = len(result.Reports) + batchNotification, err := notifyBatch(ctx, req.Config, req.Batch, batchRunID(startedAt, req.Batch), startedAt, result, plannedReports, store, req.Notifier) + if batchNotification != nil { + result.Notification = batchNotification + } + if err != nil { + result.Failed++ + } result.FinishedAt = time.Now() return result, nil } @@ -1044,6 +1074,38 @@ func (n distributorNotifier) Notify(ctx context.Context, req NotificationRequest return notification, nil } +func (n distributorNotifier) NotifyBatch(ctx context.Context, req batchNotificationRequest) (*NotificationResult, error) { + result, err := n.client.Upload(ctx, batchDistributorUploadRequest(req)) + notification := notificationResultFromUpload(req.PipelineID, req.BundleID, req.IdempotencyKey, result) + if err != nil { + return notification, err + } + return notification, nil +} + +func notificationResultFromUpload(pipelineID string, bundleID string, idempotencyKey string, result distributoradapter.UploadResult) *NotificationResult { + notification := &NotificationResult{ + PipelineID: pipelineID, + BundleID: bundleID, + IdempotencyKey: idempotencyKey, + RunID: result.RunID, + Status: result.Status, + UploadStatus: result.UploadStatus, + StatusError: result.StatusError, + } + if result.RunStatus != nil { + if result.RunStatus.PipelineID != "" { + notification.PipelineID = result.RunStatus.PipelineID + } + notification.AcceptedAt = result.RunStatus.AcceptedAt + notification.StartedAt = result.RunStatus.StartedAt + notification.FinishedAt = result.RunStatus.FinishedAt + notification.Report = append([]byte(nil), result.RunStatus.Report...) + notification.Error = result.RunStatus.Error + } + return notification +} + func distributorUploadFiles(sourcePath string, bundlePaths []string) []distributoradapter.UploadFile { files := make([]distributoradapter.UploadFile, 0, len(bundlePaths)) for _, bundlePath := range bundlePaths { diff --git a/internal/app/app_test.go b/internal/app/app_test.go index a7c7512..80a7574 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -2447,10 +2447,11 @@ func TestBuildBatchNotificationRequestRejectsMissingReportPath(t *testing.T) { func TestRunBatchContinuesAfterReportFailure(t *testing.T) { server := dailyBundleServer(t) - cfg := dailyWorkspaceConfig(t, server) + cfg := dailyNotificationConfig(t, server) collection := collectionWithFutureDailyForTest(t, cfg, "2026-05-31") cfg.WeatherAPI.BaseURL = "" collector := &recordingCollector{result: &collection} + notifier := &recordingNotifier{} renderer := &selectiveRenderer{ failRenderPrompt: "weather.tomorrow_generated_text", runBody: "# Batch Report\n", @@ -2462,6 +2463,7 @@ func TestRunBatchContinuesAfterReportFailure(t *testing.T) { Now: mustParse("2026-05-29T05:00:00-05:00"), Collector: collector, Renderer: renderer, + Notifier: notifier, }) if err != nil { t.Fatalf("RunBatchDetailed() error = %v", err) @@ -2476,6 +2478,12 @@ func TestRunBatchContinuesAfterReportFailure(t *testing.T) { if renderer.runCalls != 0 || renderer.structuredRunCalls != 2 { t.Fatalf("renderer calls run=%d structured=%d, want successful reports to continue", renderer.runCalls, renderer.structuredRunCalls) } + if len(notifier.requests) != 0 || len(notifier.batchRequests) != 0 { + t.Fatalf("notification requests report=%d batch=%d, want none after report failure", len(notifier.requests), len(notifier.batchRequests)) + } + if result.Notification == nil || result.Notification.Status != "skipped" || result.Notification.Reason != "one or more reports failed" { + t.Fatalf("batch notification = %#v, want skipped after report failure", result.Notification) + } var failedTomorrow bool var succeededDaily bool for _, item := range result.Reports { @@ -2497,6 +2505,64 @@ func TestRunBatchContinuesAfterReportFailure(t *testing.T) { } } +func TestRunBatchMorningSendsOneBatchNotification(t *testing.T) { + server := dailyBundleServer(t) + cfg := dailyNotificationConfig(t, server) + collection := collectionWithFutureDailyForTest(t, cfg, "2026-05-31") + cfg.WeatherAPI.BaseURL = "" + collector := &recordingCollector{result: &collection} + notifier := &recordingNotifier{batchResult: successfulBatchNotificationResult()} + + result, err := RunBatchDetailed(context.Background(), BatchRequest{ + Config: cfg, + Batch: BatchMorning, + Now: mustParse("2026-05-29T05:00:00-05:00"), + Collector: collector, + Renderer: &selectiveRenderer{runBody: "# Batch Report\n"}, + Notifier: notifier, + }) + if err != nil { + t.Fatalf("RunBatchDetailed() error = %v", err) + } + if result.Total != 3 || result.Succeeded != 3 || result.Failed != 0 { + t.Fatalf("summary total/succeeded/failed = %d/%d/%d, want 3/3/0", result.Total, result.Succeeded, result.Failed) + } + if len(notifier.requests) != 0 { + t.Fatalf("per-report notification requests = %#v, want none", notifier.requests) + } + if len(notifier.batchRequests) != 1 { + t.Fatalf("batch notification requests = %d, want 1", len(notifier.batchRequests)) + } + req := notifier.batchRequests[0] + if req.Batch != BatchMorning || req.RunID != "20260529T100000.000000000Z_morning" { + t.Fatalf("batch request identity = %s/%s, want morning run id", req.Batch, req.RunID) + } + if len(req.IncludedReports) != 3 || len(req.Files) != 3 { + t.Fatalf("batch request reports/files = %d/%d, want 3/3", len(req.IncludedReports), len(req.Files)) + } + for _, file := range req.Files { + if file.SourcePath == "" || file.BundlePath == "" { + t.Fatalf("batch file = %#v, want source and bundle path", file) + } + if !strings.Contains(file.BundlePath, file.RunID) { + t.Fatalf("bundle path %q does not include report run id %q", file.BundlePath, file.RunID) + } + } + if result.Notification == nil || result.Notification.Status != "succeeded" || result.Notification.RunID != "batch-distributor-run" || result.Notification.Path == "" { + t.Fatalf("batch notification = %#v, want succeeded result with artifact path", result.Notification) + } + if len(result.Notification.IncludedReports) != 3 { + t.Fatalf("batch notification included reports = %d, want 3", len(result.Notification.IncludedReports)) + } + artifact := readBatchNotificationForTest(t, result.Notification.Path) + if artifact.Status != "succeeded" || artifact.Upload == nil || artifact.Upload.RunID != "batch-distributor-run" || artifact.RunStatus == nil { + t.Fatalf("batch notification artifact = %#v, want succeeded upload and run status", artifact) + } + if len(artifact.Reports) != 3 { + t.Fatalf("artifact included reports = %d, want 3", len(artifact.Reports)) + } +} + func TestRunBatchSuppressesPerReportNotification(t *testing.T) { server := dailyBundleServer(t) cfg := dailyNotificationConfig(t, server) @@ -2528,6 +2594,9 @@ func TestRunBatchSuppressesPerReportNotification(t *testing.T) { if len(notifier.requests) != 0 { t.Fatalf("notification requests = %#v, want none for batch-generated reports", notifier.requests) } + if len(notifier.batchRequests) != 1 { + t.Fatalf("batch notification requests = %d, want 1", len(notifier.batchRequests)) + } for _, item := range result.Reports { if item.Status != "succeeded" { t.Fatalf("report %s status = %s, want succeeded", item.ReportID, item.Status) @@ -2540,11 +2609,152 @@ func TestRunBatchSuppressesPerReportNotification(t *testing.T) { t.Fatalf("report %s metadata NotificationPath = %q, want empty", item.ReportID, metadata.NotificationPath) } } - notificationDir := filepath.Join(cfg.Workspace.Root, cfg.Workspace.NotificationsDir) - if _, err := os.Stat(notificationDir); err == nil { - t.Fatalf("notification directory %q exists, want no per-report notification artifacts", notificationDir) - } else if !os.IsNotExist(err) { - t.Fatalf("stat notification directory %q: %v", notificationDir, err) + for _, item := range result.Reports { + reportNotificationDir := filepath.Join(cfg.Workspace.Root, cfg.Workspace.NotificationsDir, string(item.ReportID)) + if _, err := os.Stat(reportNotificationDir); err == nil { + t.Fatalf("per-report notification directory %q exists, want none", reportNotificationDir) + } else if !os.IsNotExist(err) { + t.Fatalf("stat per-report notification directory %q: %v", reportNotificationDir, err) + } + } +} + +func TestRunBatchNotificationFailureKeepsReportItemsSucceeded(t *testing.T) { + server := dailyBundleServer(t) + cfg := dailyNotificationConfig(t, server) + collection := collectionWithFutureDailyForTest(t, cfg, "2026-05-31") + cfg.WeatherAPI.BaseURL = "" + notifier := &recordingNotifier{batchErr: errors.New("batch upload rejected")} + + result, err := RunBatchDetailed(context.Background(), BatchRequest{ + Config: cfg, + Batch: BatchEvening, + Now: mustParse("2026-05-29T18:00:00-05:00"), + Collector: &recordingCollector{result: &collection}, + Renderer: &selectiveRenderer{runBody: "# Batch Report\n"}, + Notifier: notifier, + }) + if err != nil { + t.Fatalf("RunBatchDetailed() error = %v", err) + } + if result.Failed != 1 || result.Succeeded != 2 { + t.Fatalf("summary succeeded/failed = %d/%d, want report successes plus notification failure", result.Succeeded, result.Failed) + } + for _, item := range result.Reports { + if item.Status != "succeeded" { + t.Fatalf("report %s status = %s, want succeeded despite batch notification failure", item.ReportID, item.Status) + } + } + if result.Notification == nil || result.Notification.Status != "failed" || !strings.Contains(result.Notification.Error, "batch upload rejected") || result.Notification.Path == "" { + t.Fatalf("batch notification = %#v, want failed upload result", result.Notification) + } + artifact := readBatchNotificationForTest(t, result.Notification.Path) + if artifact.Status != "failed" || !strings.Contains(artifact.Error, "batch upload rejected") { + t.Fatalf("batch notification artifact = %#v, want failed upload error", artifact) + } + + err = RunBatch(context.Background(), BatchRequest{ + Config: cfg, + Batch: BatchEvening, + Now: mustParse("2026-05-29T18:00:00-05:00"), + Collector: &recordingCollector{result: &collection}, + Renderer: &selectiveRenderer{runBody: "# Batch Report\n"}, + Notifier: &recordingNotifier{batchErr: errors.New("batch upload rejected")}, + }) + var batchErr BatchError + if !errors.As(err, &batchErr) { + t.Fatalf("RunBatch() error = %T %v, want BatchError", err, err) + } + if batchErr.Result == nil || !batchNotificationFailed(batchErr.Result) || batchReportFailures(batchErr.Result) != 0 { + t.Fatalf("RunBatch() result = %#v, want notification-only batch failure", batchErr.Result) + } +} + +func TestRunBatchNotificationStatusErrorPersistsStatusReport(t *testing.T) { + server := dailyBundleServer(t) + cfg := dailyNotificationConfig(t, server) + collection := collectionWithFutureDailyForTest(t, cfg, "2026-05-31") + cfg.WeatherAPI.BaseURL = "" + notifier := &recordingNotifier{ + batchResult: &NotificationResult{ + RunID: "batch-distributor-run", + Status: "accepted", + UploadStatus: "accepted", + StatusError: "status lookup unavailable", + Report: []byte(`{"actions":[{"action":"replace_older"}]}`), + }, + } + + result, err := RunBatchDetailed(context.Background(), BatchRequest{ + Config: cfg, + Batch: BatchMorning, + Now: mustParse("2026-05-29T05:00:00-05:00"), + Collector: &recordingCollector{result: &collection}, + Renderer: &selectiveRenderer{runBody: "# Batch Report\n"}, + Notifier: notifier, + }) + if err != nil { + t.Fatalf("RunBatchDetailed() error = %v", err) + } + if result.Failed != 0 || result.Notification == nil || result.Notification.Path == "" { + t.Fatalf("result = %#v, want status-error notification artifact without batch failure", result) + } + artifact := readBatchNotificationForTest(t, result.Notification.Path) + if artifact.StatusError != "status lookup unavailable" || artifact.RunStatus == nil || !strings.Contains(string(artifact.RunStatus.Report), "replace_older") { + t.Fatalf("batch notification artifact = %#v, want status error and raw status report", artifact) + } +} + +func TestRunBatchDisabledDistributorSkipsBatchNotification(t *testing.T) { + server := dailyBundleServer(t) + cfg := dailyWorkspaceConfig(t, server) + collection := collectionWithFutureDailyForTest(t, cfg, "2026-05-31") + cfg.WeatherAPI.BaseURL = "" + notifier := &recordingNotifier{} + + result, err := RunBatchDetailed(context.Background(), BatchRequest{ + Config: cfg, + Batch: BatchEvening, + Now: mustParse("2026-05-29T18:00:00-05:00"), + Collector: &recordingCollector{result: &collection}, + Renderer: &selectiveRenderer{runBody: "# Batch Report\n"}, + Notifier: notifier, + }) + if err != nil { + t.Fatalf("RunBatchDetailed() error = %v", err) + } + if result.Notification != nil || result.Failed != 0 { + t.Fatalf("result notification/failed = %#v/%d, want disabled notification omitted", result.Notification, result.Failed) + } + if len(notifier.requests) != 0 || len(notifier.batchRequests) != 0 { + t.Fatalf("notification requests report=%d batch=%d, want none when distributor disabled", len(notifier.requests), len(notifier.batchRequests)) + } +} + +func TestRunBatchDisabledBatchNotificationSkipsNotifier(t *testing.T) { + server := dailyBundleServer(t) + cfg := dailyNotificationConfig(t, server) + cfg.Notify.Distributor.Batch.Enabled = false + collection := collectionWithFutureDailyForTest(t, cfg, "2026-05-31") + cfg.WeatherAPI.BaseURL = "" + notifier := &recordingNotifier{} + + result, err := RunBatchDetailed(context.Background(), BatchRequest{ + Config: cfg, + Batch: BatchEvening, + Now: mustParse("2026-05-29T18:00:00-05:00"), + Collector: &recordingCollector{result: &collection}, + Renderer: &selectiveRenderer{runBody: "# Batch Report\n"}, + Notifier: notifier, + }) + if err != nil { + t.Fatalf("RunBatchDetailed() error = %v", err) + } + if result.Notification != nil || result.Failed != 0 { + t.Fatalf("result notification/failed = %#v/%d, want disabled batch notification omitted", result.Notification, result.Failed) + } + if len(notifier.requests) != 0 || len(notifier.batchRequests) != 0 { + t.Fatalf("notification requests report=%d batch=%d, want none when batch notification disabled", len(notifier.requests), len(notifier.batchRequests)) } } @@ -2629,6 +2839,9 @@ func TestRunBatchDynamicDailyReportsHaveDistinctIdentity(t *testing.T) { if len(notifier.requests) != 0 { t.Fatalf("notification requests = %#v, want none for batch-generated reports", notifier.requests) } + if len(notifier.batchRequests) != 1 { + t.Fatalf("batch notification requests = %d, want 1", len(notifier.batchRequests)) + } dailyByDate := map[string]BatchReportResult{} runIDs := map[string]struct{}{} @@ -2675,6 +2888,18 @@ func TestRunBatchDynamicDailyReportsHaveDistinctIdentity(t *testing.T) { } } + if len(notifier.batchRequests[0].IncludedReports) != len(result.Reports) { + t.Fatalf("batch included reports = %d, want %d", len(notifier.batchRequests[0].IncludedReports), len(result.Reports)) + } + for _, included := range notifier.batchRequests[0].IncludedReports { + if strings.Contains(included.SourcePath, outputDir) { + t.Fatalf("batch notification source path = %q, want managed report path outside output dir", included.SourcePath) + } + if _, ok := reportPaths[included.SourcePath]; !ok { + t.Fatalf("batch notification source path = %q, want one of %#v", included.SourcePath, reportPaths) + } + } + } func TestRunBatchMorningUsesTodayOutputName(t *testing.T) { @@ -2842,6 +3067,15 @@ func successfulNotificationResult() *NotificationResult { } } +func successfulBatchNotificationResult() *NotificationResult { + return &NotificationResult{ + RunID: "batch-distributor-run", + Status: "succeeded", + UploadStatus: "accepted", + Report: []byte(`{"actions":[{"action":"replace_older"}]}`), + } +} + func successfulGeneratedTextRenderer(body string) *recordingRenderer { return &recordingRenderer{ renderResult: &scriptorium.RenderResult{ExitCode: 0}, @@ -3125,6 +3359,19 @@ func readMetadataForTest(t *testing.T, path string) state.Metadata { return metadata } +func readBatchNotificationForTest(t *testing.T, path string) state.BatchDistributorNotificationArtifact { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read batch notification %q: %v", path, err) + } + var artifact state.BatchDistributorNotificationArtifact + if err := json.Unmarshal(data, &artifact); err != nil { + t.Fatalf("decode batch notification %q: %v", path, err) + } + return artifact +} + func assertGeneratedReportError(t *testing.T, err error, resolved report.Resolved, operation string) { t.Helper() if err == nil { @@ -3337,10 +3584,13 @@ type selectiveRenderer struct { } type recordingNotifier struct { - requests []NotificationRequest - result *NotificationResult - err error - errByReport map[report.ID]error + requests []NotificationRequest + batchRequests []batchNotificationRequest + result *NotificationResult + batchResult *NotificationResult + err error + batchErr error + errByReport map[report.ID]error } func (n *recordingNotifier) Notify(_ context.Context, req NotificationRequest) (*NotificationResult, error) { @@ -3373,6 +3623,34 @@ func (n *recordingNotifier) Notify(_ context.Context, req NotificationRequest) ( }, nil } +func (n *recordingNotifier) NotifyBatch(_ context.Context, req batchNotificationRequest) (*NotificationResult, error) { + n.batchRequests = append(n.batchRequests, req) + if n.batchErr != nil { + return nil, n.batchErr + } + if n.batchResult != nil { + result := *n.batchResult + if result.BundleID == "" { + result.BundleID = req.BundleID + } + 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, + RunID: "batch-distributor-run", + Status: "accepted", + UploadStatus: "accepted", + }, nil +} + func (r *selectiveRenderer) Render(_ context.Context, req scriptorium.RenderRequest) (*scriptorium.RenderResult, error) { r.renderCalls++ if req.PromptID == r.failRenderPrompt { diff --git a/internal/app/batch_notification.go b/internal/app/batch_notification.go index 1815d82..9182972 100644 --- a/internal/app/batch_notification.go +++ b/internal/app/batch_notification.go @@ -1,12 +1,14 @@ package app import ( + "context" "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/state" "gitea.maximumdirect.net/eric/weatherreporter/internal/timeutil" ) @@ -36,10 +38,80 @@ type batchNotificationFile struct { BundlePath string } +type batchNotifier interface { + NotifyBatch(context.Context, batchNotificationRequest) (*NotificationResult, error) +} + func batchRunID(startedAt time.Time, batch BatchKind) string { return startedAt.UTC().Format(runIDTimestampLayout) + "_" + string(batch) } +func notifyBatch(ctx context.Context, cfg config.Config, batch BatchKind, runID string, startedAt time.Time, result *BatchResult, planned []plannedBatchReport, store state.Store, notifier Notifier) (*BatchNotificationResult, error) { + if !cfg.Notify.Distributor.Enabled { + return nil, nil + } + if !cfg.Notify.Distributor.Batch.Enabled { + return nil, nil + } + if result == nil { + return nil, fmt.Errorf("batch result is required") + } + if result.Failed > 0 { + return &BatchNotificationResult{ + Status: "skipped", + Reason: "one or more reports failed", + }, nil + } + + req, err := buildBatchNotificationRequest(cfg, batch, runID, startedAt, result.Reports, planned) + if err != nil { + path, saveErr := saveBatchNotificationArtifact(ctx, store, cfg, batch, runID, startedAt, batchNotificationRequest{}, nil, err) + if saveErr != nil { + return nil, saveErr + } + return failedBatchNotificationResult(batchNotificationRequest{}, path, err), err + } + + batchNotifier, err := resolveBatchNotifier(cfg, notifier) + if err != nil { + path, saveErr := saveBatchNotificationArtifact(ctx, store, cfg, batch, runID, startedAt, req, nil, err) + if saveErr != nil { + return nil, saveErr + } + return failedBatchNotificationResult(req, path, err), err + } + + notification, notifyErr := batchNotifier.NotifyBatch(ctx, req) + wrappedErr := notifyErr + if notifyErr != nil { + wrappedErr = fmt.Errorf("notify batch %q run %q bundle %q: %w", batch, runID, req.BundleID, notifyErr) + } + path, saveErr := saveBatchNotificationArtifact(ctx, store, cfg, batch, runID, startedAt, req, notification, wrappedErr) + if saveErr != nil { + return nil, saveErr + } + + batchResult := batchNotificationResult(req, notification, path) + if wrappedErr != nil { + batchResult.Status = "failed" + batchResult.Error = wrappedErr.Error() + return batchResult, wrappedErr + } + return batchResult, nil +} + +func resolveBatchNotifier(cfg config.Config, notifier Notifier) (batchNotifier, error) { + if notifier != nil { + if batchNotifier, ok := notifier.(batchNotifier); ok { + return batchNotifier, nil + } + return nil, fmt.Errorf("batch distributor notifier is required") + } + return distributorNotifier{ + client: distributoradapter.New(cfg.Notify.Distributor), + }, nil +} + 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") @@ -153,6 +225,118 @@ func batchDistributorUploadRequest(req batchNotificationRequest) distributoradap } } +func batchNotificationResult(req batchNotificationRequest, result *NotificationResult, path string) *BatchNotificationResult { + notification := &BatchNotificationResult{ + Status: "unknown", + PipelineID: req.PipelineID, + BundleID: req.BundleID, + IdempotencyKey: req.IdempotencyKey, + Path: path, + IncludedReports: append([]BatchNotificationReport(nil), req.IncludedReports...), + } + if result != nil { + notification.Status = result.Status + notification.RunID = result.RunID + if result.PipelineID != "" { + notification.PipelineID = result.PipelineID + } + if result.BundleID != "" { + notification.BundleID = result.BundleID + } + if result.IdempotencyKey != "" { + notification.IdempotencyKey = result.IdempotencyKey + } + if result.Error != "" { + notification.Error = result.Error + } + } + if notification.Status == "" { + notification.Status = "unknown" + } + return notification +} + +func failedBatchNotificationResult(req batchNotificationRequest, path string, err error) *BatchNotificationResult { + notification := batchNotificationResult(req, nil, path) + notification.Status = "failed" + if err != nil { + notification.Error = err.Error() + } + return notification +} + +func saveBatchNotificationArtifact(ctx context.Context, store state.Store, cfg config.Config, batch BatchKind, runID string, startedAt time.Time, req batchNotificationRequest, result *NotificationResult, notifyErr error) (string, error) { + if store == nil { + return "", fmt.Errorf("state store is required") + } + location, err := timeutil.LoadLocation(cfg.WeatherAPI.Timezone) + if err != nil { + return "", fmt.Errorf("load batch notification timezone: %w", err) + } + artifact := state.BatchDistributorNotificationArtifact{ + SchemaVersion: state.BatchDistributorNotificationSchemaVersion, + Batch: string(batch), + BatchRunID: runID, + AttemptedAt: time.Now(), + Endpoint: cfg.Notify.Distributor.Endpoint, + PipelineID: req.PipelineID, + BundleID: req.BundleID, + IdempotencyKey: req.IdempotencyKey, + BundleCreated: req.CreatedAt, + Reports: batchNotificationReportArtifacts(req.IncludedReports), + Status: "attempted", + } + if result != nil { + artifact.Status = result.Status + artifact.Upload = &state.DistributorUploadResult{ + RunID: result.RunID, + Status: result.UploadStatus, + } + if result.PipelineID != "" || !result.AcceptedAt.IsZero() || result.StartedAt != nil || result.FinishedAt != nil || len(result.Report) > 0 || result.Error != "" { + artifact.RunStatus = &state.DistributorRunStatus{ + RunID: result.RunID, + PipelineID: result.PipelineID, + Status: result.Status, + AcceptedAt: result.AcceptedAt, + StartedAt: result.StartedAt, + FinishedAt: result.FinishedAt, + Report: append([]byte(nil), result.Report...), + Error: result.Error, + } + } + artifact.StatusError = result.StatusError + } + if notifyErr != nil { + artifact.Status = "failed" + artifact.Error = notifyErr.Error() + } + if artifact.Status == "" { + artifact.Status = "unknown" + } + return store.SaveBatchDistributorNotification(ctx, state.BatchDistributorNotificationRef{ + Batch: string(batch), + BatchRunID: runID, + StartedAt: startedAt, + Location: location, + }, artifact) +} + +func batchNotificationReportArtifacts(reports []BatchNotificationReport) []state.BatchDistributorNotificationReportArtifact { + if len(reports) == 0 { + return nil + } + artifacts := make([]state.BatchDistributorNotificationReportArtifact, 0, len(reports)) + for _, item := range reports { + artifacts = append(artifacts, state.BatchDistributorNotificationReportArtifact{ + ReportID: item.ReportID, + RunID: item.RunID, + SourcePath: item.SourcePath, + BundlePaths: append([]string(nil), item.BundlePaths...), + }) + } + return artifacts +} + func renderBatchNotificationIdentity(cfg config.Config, batch BatchKind, runID string, startedAt time.Time) (batchNotificationIdentity, error) { values, err := batchNotificationTemplateValues(cfg, batch, runID, startedAt) if err != nil { diff --git a/internal/cli/root_test.go b/internal/cli/root_test.go index d5a4668..4418225 100644 --- a/internal/cli/root_test.go +++ b/internal/cli/root_test.go @@ -323,8 +323,20 @@ func TestRunEveningUsesOutputDirectoryAndSummary(t *testing.T) { func TestRunEveningReportsOmitsPerReportNotification(t *testing.T) { server := dailyServer(t) + var uploadCount int distributorServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - t.Fatalf("unexpected distributor request %s", r.URL.Path) + if r.URL.Path == "/runs/batch-distributor-run" { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"run_id":"batch-distributor-run","pipeline_id":"weatherreporter","status":"succeeded","report":{"actions":[{"action":"replace_older"}]}}`)) + return + } + if r.URL.Path != "/v1/pipelines/weatherreporter/upload" { + http.NotFound(w, r) + return + } + uploadCount++ + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`{"run_id":"batch-distributor-run","status":"accepted"}`)) })) t.Cleanup(distributorServer.Close) tempDir := t.TempDir() @@ -351,6 +363,12 @@ func TestRunEveningReportsOmitsPerReportNotification(t *testing.T) { if len(summary.Reports) != 1 { t.Fatalf("reports = %#v, want one report", summary.Reports) } + if uploadCount != 1 { + t.Fatalf("batch upload count = %d, want 1", uploadCount) + } + if summary.Notification == nil || summary.Notification.Status != "succeeded" || summary.Notification.RunID != "batch-distributor-run" { + t.Fatalf("batch notification = %#v, want succeeded batch notification", summary.Notification) + } if summary.Reports[0].NotificationStatus != "" || summary.Reports[0].NotificationRunID != "" || summary.Reports[0].NotificationPipelineID != "" || summary.Reports[0].NotificationError != "" || summary.Reports[0].NotificationPath != "" { t.Fatalf("notification fields = %#v, want empty per-report notification fields", summary.Reports[0]) } @@ -371,7 +389,7 @@ func TestRunEveningReportsDoesNotRequirePerReportDistributorToken(t *testing.T) tempDir := t.TempDir() scriptoriumPath := writeFakeScriptorium(t, tempDir) workspaceRoot := filepath.Join(tempDir, "workspace") - configPath := writeTestConfigWithDistributor(t, server, scriptoriumPath, workspaceRoot, distributorServer.URL) + configPath := writeTestConfigWithDisabledBatchDistributor(t, server, scriptoriumPath, workspaceRoot, distributorServer.URL) var stdout bytes.Buffer var stderr bytes.Buffer runner := Runner{Clock: fixedClock()} @@ -391,6 +409,9 @@ func TestRunEveningReportsDoesNotRequirePerReportDistributorToken(t *testing.T) if len(summary.Reports) != 1 { t.Fatalf("summary reports = %#v, want one report", summary.Reports) } + if summary.Notification != nil { + t.Fatalf("batch notification = %#v, want omitted when disabled", summary.Notification) + } if summary.Reports[0].NotificationStatus != "" || summary.Reports[0].NotificationError != "" { t.Fatalf("notification fields = %#v, want empty per-report notification fields", summary.Reports[0]) } @@ -1144,6 +1165,12 @@ func writeTestConfigWithDistributor(t *testing.T, server *httptest.Server, scrip return writeConfigFile(t, configBody) } +func writeTestConfigWithDisabledBatchDistributor(t *testing.T, server *httptest.Server, scriptoriumPath string, workspaceRoot string, distributorEndpoint string) string { + t.Helper() + 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: " + distributorEndpoint + "\n token_env: CLI_DISTRIBUTOR_TOKEN\n pipeline_id_template: weatherreporter.{artifact_group}\n batch:\n enabled: false\n" + return writeConfigFile(t, configBody) +} + func writeWorkspaceConfig(t *testing.T, workspaceRoot string) string { t.Helper() return writeConfigFile(t, "workspace:\n root: "+workspaceRoot+"\n")