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" ) 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, 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") } 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 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 { 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 }