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

Build batch distributor upload requests

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-17 20:48:38 +00:00
parent a82f03feb8
commit 2b3bcdd4f1
3 changed files with 288 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
internal/app/app.go
View File

@@ -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 {

136
internal/app/app_test.go
View File

@@ -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})

133
internal/app/batch_notification.go
View File

@@ -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 {