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

Stop persisting notification receipts

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-01 19:40:51 +00:00
parent b184ca7cbd
commit 4bdba6f2b7
9 changed files with 94 additions and 310 deletions
Download Patch File Download Diff File Expand all files Collapse all files

148
internal/app/app.go
View File

@@ -94,7 +94,6 @@ type ReportResult struct {
LLMDebugPath string
ReportPath string
OutputPath string
NotificationPath string
Metadata state.Metadata
MetadataPath string
GeneratedTextRawPath string
@@ -121,7 +120,6 @@ type BatchNotificationResult struct {
PipelineID string `json:"pipelineId,omitempty"`
BundleID string `json:"bundleId,omitempty"`
IdempotencyKey string `json:"idempotencyKey,omitempty"`
Path string `json:"path,omitempty"`
IncludedReports []BatchNotificationReport `json:"includedReports,omitempty"`
Error string `json:"error,omitempty"`
}
@@ -144,7 +142,6 @@ type BatchReportResult struct {
NotificationRunID string `json:"notificationRunId,omitempty"`
NotificationPipelineID string `json:"notificationPipelineId,omitempty"`
NotificationError string `json:"notificationError,omitempty"`
NotificationPath string `json:"notificationPath,omitempty"`
GeneratedAt time.Time `json:"generatedAt"`
ValidPeriod timeutil.Period `json:"validPeriod"`
DataPackagePath string `json:"dataPackagePath,omitempty"`
@@ -387,7 +384,7 @@ 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)
batchNotification, err := notifyBatch(ctx, req.Config, req.Batch, batchRunID(startedAt, req.Batch), startedAt, result, plannedReports, req.Notifier)
if batchNotification != nil {
result.Notification = batchNotification
}
@@ -408,7 +405,6 @@ func copyBatchReportPaths(item *BatchReportResult, result *ReportResult) {
item.ReportPath = result.ReportPath
item.OutputPath = result.OutputPath
item.MetadataPath = result.MetadataPath
item.NotificationPath = result.NotificationPath
if result.Notification != nil {
item.NotificationStatus = result.Notification.Status
item.NotificationRunID = result.Notification.RunID
@@ -617,33 +613,32 @@ func FetchAndSaveBundle(ctx context.Context, req FetchBundleRequest) (*weatherda
}
type finalizeRenderedReportRequest struct {
Config config.Config
Store state.Store
Resolved report.Resolved
Metadata state.Metadata
MetadataPath string
ExecutionArtifact *state.PromptExecutionArtifact
ManagedReportPath string
OutputPath string
Notifier Notifier
GenerationErr error
noNotify bool
Config config.Config
Store state.Store
Resolved report.Resolved
Metadata state.Metadata
MetadataPath string
ExecutionArtifact *state.PromptExecutionArtifact
RenderedReportPath string
OutputPath string
Notifier Notifier
GenerationErr error
noNotify bool
}
type finalizeRenderedReportResult struct {
OutputPath string
NotificationPath string
Metadata state.Metadata
MetadataPath string
Notification *NotificationResult
OutputPath string
Metadata state.Metadata
MetadataPath string
Notification *NotificationResult
}
func finalizeRenderedReport(ctx context.Context, req finalizeRenderedReportRequest) (finalizeRenderedReportResult, error) {
if req.Store == nil {
return finalizeRenderedReportResult{}, fmt.Errorf("state store is required")
}
if req.ManagedReportPath == "" {
return finalizeRenderedReportResult{}, fmt.Errorf("managed report path is required for report %q", req.Resolved.Definition.ID)
if req.RenderedReportPath == "" {
return finalizeRenderedReportResult{}, fmt.Errorf("rendered report path is required for report %q", req.Resolved.Definition.ID)
}
if req.ExecutionArtifact == nil {
return finalizeRenderedReportResult{}, fmt.Errorf("prompt execution artifact is required for report %q", req.Resolved.Definition.ID)
@@ -651,8 +646,8 @@ func finalizeRenderedReport(ctx context.Context, req finalizeRenderedReportReque
result := finalizeRenderedReportResult{Metadata: req.Metadata, MetadataPath: req.MetadataPath}
if req.OutputPath != "" && req.GenerationErr == nil {
if req.OutputPath != req.ManagedReportPath {
if err := fileutil.CopyFileAtomic(req.ManagedReportPath, req.OutputPath); err != nil {
if req.OutputPath != req.RenderedReportPath {
if err := fileutil.CopyFileAtomic(req.RenderedReportPath, req.OutputPath); err != nil {
return result, err
}
result.OutputPath = req.OutputPath
@@ -667,7 +662,7 @@ func finalizeRenderedReport(ctx context.Context, req finalizeRenderedReportReque
}
metadata := req.Metadata
metadata.RenderedReportPath = req.ManagedReportPath
metadata.RenderedReportPath = req.RenderedReportPath
metadataPath, err := req.Store.SaveMetadata(ctx, metadata)
if err != nil {
return result, err
@@ -681,24 +676,7 @@ func finalizeRenderedReport(ctx context.Context, req finalizeRenderedReportReque
return result, nil
}
notification, notificationPath, err := notifyReport(ctx, req.Config, req.Resolved, req.OutputPath, metadata, req.Notifier, req.Store)
if notificationPath != "" {
result.NotificationPath = notificationPath
result.Notification = notification
metadata.NotificationPath = notificationPath
result.Metadata = metadata
if saveErr := persistReachedPromptPath(ctx, req.Store, req.Resolved, req.ExecutionArtifact, func(paths *state.PromptExecutionPaths) {
paths.NotificationPath = notificationPath
}); saveErr != nil {
return result, saveErr
}
metadataPath, saveErr := req.Store.SaveMetadata(ctx, metadata)
if saveErr != nil {
return result, saveErr
}
result.Metadata = metadata
result.MetadataPath = metadataPath
}
notification, err := notifyReport(ctx, req.Config, req.Resolved, req.OutputPath, metadata.RunID, metadata.GeneratedAt, req.Notifier)
result.Notification = notification
if err != nil {
return result, err
@@ -706,31 +684,23 @@ func finalizeRenderedReport(ctx context.Context, req finalizeRenderedReportReque
return result, nil
}
func notifyReport(ctx context.Context, cfg config.Config, resolved report.Resolved, reportPath string, metadata state.Metadata, notifier Notifier, store state.Store) (*NotificationResult, string, error) {
func notifyReport(ctx context.Context, cfg config.Config, resolved report.Resolved, outputPath, runID string, generatedAt time.Time, notifier Notifier) (*NotificationResult, error) {
notifier, enabled := reportNotifier(cfg, notifier)
if !enabled {
return nil, "", nil
return nil, nil
}
notificationRequest, err := buildNotificationRequest(cfg, resolved, reportPath, metadata)
notificationRequest, err := buildNotificationRequest(cfg, resolved, outputPath, runID, generatedAt)
if err != nil {
notificationPath, saveErr := saveNotificationArtifact(ctx, store, resolved, cfg, metadata, NotificationRequest{}, nil, err)
if saveErr != nil {
return nil, "", saveErr
}
return nil, notificationPath, err
return nil, err
}
result, err := notifier.Notify(ctx, notificationRequest)
notificationPath, saveErr := saveNotificationArtifact(ctx, store, resolved, cfg, metadata, notificationRequest, result, err)
if saveErr != nil {
return nil, "", saveErr
}
if err != nil {
return result, notificationPath, &NotificationError{
return result, &NotificationError{
Request: notificationRequest,
Err: fmt.Errorf("notify report %q run %q from output %q: %w", resolved.Definition.ID, metadata.RunID, reportPath, err),
Err: fmt.Errorf("notify report %q run %q from output %q: %w", resolved.Definition.ID, runID, outputPath, err),
}
}
return result, notificationPath, nil
return result, nil
}
func reportNotifier(cfg config.Config, notifier Notifier) (Notifier, bool) {
@@ -745,8 +715,8 @@ func reportNotifier(cfg config.Config, notifier Notifier) (Notifier, bool) {
}, true
}
func buildNotificationRequest(cfg config.Config, resolved report.Resolved, reportPath string, metadata state.Metadata) (NotificationRequest, error) {
values, err := distributorTemplateValuesForReport(cfg, resolved, metadata.RunID, filepath.Base(reportPath))
func buildNotificationRequest(cfg config.Config, resolved report.Resolved, outputPath, runID string, generatedAt time.Time) (NotificationRequest, error) {
values, err := distributorTemplateValuesForReport(cfg, resolved, runID, filepath.Base(outputPath))
if err != nil {
return NotificationRequest{}, err
}
@@ -763,19 +733,19 @@ func buildNotificationRequest(cfg config.Config, resolved report.Resolved, repor
if err != nil {
return NotificationRequest{}, err
}
bundlePaths, err := renderDistributorReportBundlePaths(cfg, resolved, metadata.RunID, reportPath, values)
bundlePaths, err := renderDistributorReportBundlePaths(cfg, resolved, runID, outputPath, values)
if err != nil {
return NotificationRequest{}, err
}
return NotificationRequest{
ReportID: resolved.Definition.ID,
RunID: metadata.RunID,
RunID: runID,
PipelineID: pipelineID,
BundleID: bundleID,
IdempotencyKey: idempotencyKey,
ReportPath: reportPath,
ReportPath: outputPath,
BundlePaths: bundlePaths,
CreatedAt: metadata.GeneratedAt,
CreatedAt: generatedAt,
}, nil
}
@@ -849,54 +819,6 @@ func addDistributorValidPeriodValues(values *config.DistributorTemplateValues, p
return nil
}
func saveNotificationArtifact(ctx context.Context, store state.Store, resolved report.Resolved, cfg config.Config, metadata state.Metadata, req NotificationRequest, result *NotificationResult, notifyErr error) (string, error) {
if store == nil {
return "", fmt.Errorf("state store is required")
}
artifact := state.DistributorNotificationArtifact{
SchemaVersion: state.DistributorNotificationSchemaVersion,
RunID: metadata.RunID,
ReportID: resolved.Definition.ID,
AttemptedAt: time.Now(),
Endpoint: cfg.Notify.Distributor.Endpoint,
PipelineID: req.PipelineID,
BundleID: req.BundleID,
IdempotencyKey: req.IdempotencyKey,
SourcePath: req.ReportPath,
BundlePaths: append([]string(nil), req.BundlePaths...),
BundleCreated: req.CreatedAt,
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.SaveDistributorNotification(ctx, resolved, artifact)
}
type noopNotifier struct{}
func (noopNotifier) Notify(context.Context, NotificationRequest) (*NotificationResult, error) {