package app import ( "context" "fmt" "path/filepath" "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" ) const runIDTimestampLayout = "20060102T150405.000000000Z" type batchNotificationIdentity struct { PipelineID string BundleID string 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 } 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, notifier Notifier) *BatchNotificationResult { if !cfg.Notify.Distributor.Enabled { return nil } if !cfg.Notify.Distributor.Batch.Enabled { return nil } if result == nil { return failedBatchNotificationResult(batchNotificationRequest{}, fmt.Errorf("batch result is required")) } if result.Canceled > 0 { return &BatchNotificationResult{ Status: "skipped", Reason: "batch canceled", } } if result.Failed > 0 { return &BatchNotificationResult{ Status: "skipped", Reason: "one or more reports failed", } } req, err := buildBatchNotificationRequest(cfg, batch, runID, startedAt, result.Reports, planned) if err != nil { return failedBatchNotificationResult(batchNotificationRequest{}, err) } batchNotifier, err := resolveBatchNotifier(cfg, notifier) if err != nil { return failedBatchNotificationResult(req, 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) } batchResult := batchNotificationResult(req, notification) if wrappedErr != nil { batchResult.Status = "failed" batchResult.Error = safeDistributorNotificationFailure(wrappedErr) return batchResult } return batchResult } 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") } 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.OutputPath == "" { return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q is missing output path", item.ReportID, item.RunID) } values, err := distributorTemplateValuesForReport(cfg, plannedReport.Resolved, item.RunID, filepath.Base(item.OutputPath)) if err != nil { return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q source path %q: %w", item.ReportID, item.RunID, item.OutputPath, err) } bundlePaths, err := renderDistributorReportBundlePaths(cfg, plannedReport.Resolved, item.RunID, item.OutputPath, values) if err != nil { return batchNotificationRequest{}, err } included := BatchNotificationReport{ ReportID: item.ReportID, RunID: item.RunID, SourcePath: item.OutputPath, BundlePaths: append([]string(nil), bundlePaths...), } for _, bundlePath := range bundlePaths { file := batchNotificationFile{ ReportID: item.ReportID, RunID: item.RunID, SourcePath: item.OutputPath, 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.OutputPath, 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 batchNotificationResult(req batchNotificationRequest, result *NotificationResult) *BatchNotificationResult { notification := &BatchNotificationResult{ Status: "unknown", PipelineID: req.PipelineID, BundleID: req.BundleID, IdempotencyKey: req.IdempotencyKey, 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 = safeDistributorRunError(result.Error) } } if notification.Status == "" { notification.Status = "unknown" } return notification } func failedBatchNotificationResult(req batchNotificationRequest, err error) *BatchNotificationResult { notification := batchNotificationResult(req, nil) notification.Status = "failed" if err != nil { notification.Error = safeDistributorNotificationFailure(err) } return notification } func safeDistributorNotificationFailure(err error) string { if err == nil { return "" } return "distributor notification failed" } func renderBatchNotificationIdentity(cfg config.Config, batch BatchKind, runID string, startedAt time.Time) (batchNotificationIdentity, error) { values, err := batchNotificationTemplateValues(cfg, batch, runID, startedAt) if err != nil { return batchNotificationIdentity{}, err } bundleID, err := config.RenderDistributorBatchBundleID(cfg.Notify.Distributor.Batch.BundleIDTemplate, values) if err != nil { return batchNotificationIdentity{}, err } values.BundleID = bundleID pipelineID, err := config.RenderDistributorBatchPipelineID(cfg.Notify.Distributor.Batch.PipelineIDTemplate, values) if err != nil { return batchNotificationIdentity{}, err } idempotencyKey, err := config.RenderDistributorBatchIdempotencyKey(cfg.Notify.Distributor.Batch.IdempotencyKeyTemplate, values) if err != nil { return batchNotificationIdentity{}, err } return batchNotificationIdentity{ PipelineID: pipelineID, BundleID: bundleID, IdempotencyKey: idempotencyKey, }, nil } func batchNotificationTemplateValues(cfg config.Config, batch BatchKind, runID string, startedAt time.Time) (config.DistributorBatchTemplateValues, error) { location, err := timeutil.LoadLocation(cfg.WeatherAPI.Timezone) if err != nil { return config.DistributorBatchTemplateValues{}, fmt.Errorf("load batch notification timezone: %w", err) } return config.DistributorBatchTemplateValues{ LocationID: cfg.Location.ID, Batch: string(batch), BatchRunID: runID, BatchStartedDate: startedAt.In(location).Format(timeutil.DateLayout), }, nil }