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

Add batch distributor notification state artifacts

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-17 20:39:22 +00:00
parent 32060bd370
commit f1d4e38414
3 changed files with 286 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
internal/state/filesystem.go
View File

@@ -141,6 +141,33 @@ func (s *FilesystemStore) SaveDistributorNotification(_ context.Context, resolve
}, artifact)
}
func (s *FilesystemStore) BatchDistributorNotificationPath(ref BatchDistributorNotificationRef) (string, error) {
if s == nil {
return "", fmt.Errorf("state store is required")
}
if err := validateBatchNotificationRef(ref); err != nil {
return "", err
}
localDate := ref.StartedAt.In(ref.Location).Format("2006-01-02")
return s.join(s.notificationsDir, "batches", ref.Batch, localDate, ref.BatchRunID+".distributor.json"), nil
}
func (s *FilesystemStore) SaveBatchDistributorNotification(_ context.Context, ref BatchDistributorNotificationRef, artifact BatchDistributorNotificationArtifact) (string, error) {
path, err := s.BatchDistributorNotificationPath(ref)
if err != nil {
return "", err
}
if artifact.SchemaVersion == "" {
artifact.SchemaVersion = BatchDistributorNotificationSchemaVersion
}
artifact.Batch = ref.Batch
artifact.BatchRunID = ref.BatchRunID
if err := fileutil.WriteJSONAtomic(path, artifact); err != nil {
return "", err
}
return path, nil
}
func (s *FilesystemStore) SaveGeneratedTextRaw(_ context.Context, resolved report.Resolved, data []byte) (string, error) {
return s.saveResolvedBytes(resolved, func(paths ArtifactPaths) string {
return paths.GeneratedTextRaw
@@ -461,6 +488,35 @@ func validateRelativeDir(name string, value string) error {
return nil
}
func validateBatchNotificationRef(ref BatchDistributorNotificationRef) error {
if err := validatePathSegment("batch kind", ref.Batch); err != nil {
return err
}
if err := validatePathSegment("batch run id", ref.BatchRunID); err != nil {
return err
}
if ref.StartedAt.IsZero() {
return fmt.Errorf("batch started time is required")
}
if ref.Location == nil {
return fmt.Errorf("batch location is required")
}
return nil
}
func validatePathSegment(name string, value string) error {
if strings.TrimSpace(value) == "" {
return fmt.Errorf("%s is required", name)
}
if strings.ContainsAny(value, `/\`) {
return fmt.Errorf("%s must not contain path separators", name)
}
if value == "." || value == ".." {
return fmt.Errorf("%s must be a safe path segment", name)
}
return nil
}
func readJSON(path string, target any) error {
data, err := os.ReadFile(path)
if err != nil {

196
internal/state/filesystem_test.go
View File

@@ -69,6 +69,193 @@ func TestDailyPathsUseRunIDValidDateDisambiguator(t *testing.T) {
}
}
func TestBatchDistributorNotificationPathUsesWorkspaceBatchDateAndRunID(t *testing.T) {
store := newTestStore(t)
location := mustLoadStateLocation(t, "America/Chicago")
startedAt := time.Date(2026, 6, 18, 3, 30, 0, 123456789, time.UTC)
path, err := store.BatchDistributorNotificationPath(BatchDistributorNotificationRef{
Batch: "evening",
BatchRunID: "20260618T033000.123456789Z_evening",
StartedAt: startedAt,
Location: location,
})
if err != nil {
t.Fatalf("BatchDistributorNotificationPath() error = %v", err)
}
want := filepath.Join("notifications", "batches", "evening", "2026-06-17", "20260618T033000.123456789Z_evening.distributor.json")
if !strings.Contains(path, want) {
t.Fatalf("path = %q, want component %q", path, want)
}
if !strings.HasPrefix(path, store.root) {
t.Fatalf("path = %q, want workspace root prefix %q", path, store.root)
}
}
func TestSaveBatchDistributorNotificationRoundTrip(t *testing.T) {
store := newTestStore(t)
location := mustLoadStateLocation(t, "America/Chicago")
startedAt := time.Date(2026, 6, 17, 12, 0, 0, 0, time.UTC)
bundleCreated := startedAt.Add(2 * time.Second)
attemptedAt := startedAt.Add(3 * time.Second)
acceptedAt := startedAt.Add(4 * time.Second)
finishedAt := startedAt.Add(5 * time.Second)
ref := BatchDistributorNotificationRef{
Batch: "morning",
BatchRunID: "20260617T120000.000000000Z_morning",
StartedAt: startedAt,
Location: location,
}
path, err := store.SaveBatchDistributorNotification(context.Background(), ref, BatchDistributorNotificationArtifact{
AttemptedAt: attemptedAt,
Endpoint: "https://distributor.example.test",
PipelineID: "weatherreporter",
BundleID: "weatherreporter.home.morning",
IdempotencyKey: "weatherreporter.home.morning.20260617T120000.000000000Z_morning",
BundleCreated: bundleCreated,
Reports: []BatchDistributorNotificationReportArtifact{
{
ReportID: report.Today,
RunID: "20260617T120000.000000000Z_today",
SourcePath: "/workspace/reports/today/20260617T120000.000000000Z_today.md",
BundlePaths: []string{"2026-06-17/today/report.md"},
},
{
ReportID: report.Daily,
RunID: "20260617T120000.000000000Z_daily_2026-06-19",
SourcePath: "/workspace/reports/daily/20260617T120000.000000000Z_daily_2026-06-19.md",
BundlePaths: []string{"2026-06-19/daily/report.md"},
},
},
Status: "failed",
Upload: &DistributorUploadResult{
RunID: "distributor-run",
Status: "accepted",
},
RunStatus: &DistributorRunStatus{
RunID: "distributor-run",
PipelineID: "weatherreporter",
Status: "failed",
AcceptedAt: acceptedAt,
FinishedAt: &finishedAt,
Report: json.RawMessage(`{"actions":[{"action":"failed"}]}`),
Error: "destination conflict",
},
StatusError: "status lookup failed",
Error: "batch upload failed",
})
if err != nil {
t.Fatalf("SaveBatchDistributorNotification() error = %v", err)
}
wantPath := filepath.Join("notifications", "batches", "morning", "2026-06-17", "20260617T120000.000000000Z_morning.distributor.json")
if !strings.Contains(path, wantPath) {
t.Fatalf("path = %q, want component %q", path, wantPath)
}
data, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read batch notification: %v", err)
}
var artifact BatchDistributorNotificationArtifact
if err := json.Unmarshal(data, &artifact); err != nil {
t.Fatalf("decode batch notification: %v", err)
}
if artifact.SchemaVersion != BatchDistributorNotificationSchemaVersion {
t.Fatalf("SchemaVersion = %q, want %q", artifact.SchemaVersion, BatchDistributorNotificationSchemaVersion)
}
if artifact.Batch != "morning" || artifact.BatchRunID != ref.BatchRunID {
t.Fatalf("artifact batch identity = %q/%q, want ref values", artifact.Batch, artifact.BatchRunID)
}
if artifact.Endpoint != "https://distributor.example.test" || artifact.PipelineID != "weatherreporter" || artifact.BundleID != "weatherreporter.home.morning" || artifact.IdempotencyKey == "" {
t.Fatalf("artifact identity = %#v, want distributor identity", artifact)
}
if len(artifact.Reports) != 2 || artifact.Reports[0].ReportID != report.Today || strings.Join(artifact.Reports[1].BundlePaths, ",") != "2026-06-19/daily/report.md" {
t.Fatalf("Reports = %#v, want included report records", artifact.Reports)
}
if artifact.Upload == nil || artifact.Upload.RunID != "distributor-run" {
t.Fatalf("Upload = %#v, want accepted upload result", artifact.Upload)
}
if artifact.RunStatus == nil || artifact.RunStatus.Status != "failed" || !strings.Contains(string(artifact.RunStatus.Report), "failed") || artifact.RunStatus.FinishedAt == nil {
t.Fatalf("RunStatus = %#v, want failed run status with raw report", artifact.RunStatus)
}
if artifact.StatusError != "status lookup failed" || artifact.Error != "batch upload failed" {
t.Fatalf("errors = %q/%q, want persisted error fields", artifact.StatusError, artifact.Error)
}
}
func TestBatchDistributorNotificationPathRejectsInvalidIdentity(t *testing.T) {
store := newTestStore(t)
location := mustLoadStateLocation(t, "America/Chicago")
valid := BatchDistributorNotificationRef{
Batch: "morning",
BatchRunID: "20260617T120000.000000000Z_morning",
StartedAt: time.Date(2026, 6, 17, 12, 0, 0, 0, time.UTC),
Location: location,
}
tests := []struct {
name string
mutate func(*BatchDistributorNotificationRef)
wantErr string
}{
{
name: "Batch",
mutate: func(ref *BatchDistributorNotificationRef) {
ref.Batch = ""
},
wantErr: "batch kind is required",
},
{
name: "BatchSeparator",
mutate: func(ref *BatchDistributorNotificationRef) {
ref.Batch = "../morning"
},
wantErr: "batch kind must not contain path separators",
},
{
name: "BatchRunID",
mutate: func(ref *BatchDistributorNotificationRef) {
ref.BatchRunID = ""
},
wantErr: "batch run id is required",
},
{
name: "BatchRunIDSeparator",
mutate: func(ref *BatchDistributorNotificationRef) {
ref.BatchRunID = "nested/run"
},
wantErr: "batch run id must not contain path separators",
},
{
name: "StartedAt",
mutate: func(ref *BatchDistributorNotificationRef) {
ref.StartedAt = time.Time{}
},
wantErr: "batch started time is required",
},
{
name: "Location",
mutate: func(ref *BatchDistributorNotificationRef) {
ref.Location = nil
},
wantErr: "batch location is required",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
ref := valid
tt.mutate(&ref)
_, err := store.BatchDistributorNotificationPath(ref)
if err == nil {
t.Fatal("BatchDistributorNotificationPath() error = nil, want error")
}
if !strings.Contains(err.Error(), tt.wantErr) {
t.Fatalf("error = %q, want %q", err.Error(), tt.wantErr)
}
})
}
}
func TestGeneratedTextArtifactPathsUseSnapshotTree(t *testing.T) {
store := newTestStore(t)
tests := []struct {
@@ -595,6 +782,15 @@ func newTestStore(t *testing.T) *FilesystemStore {
return store
}
func mustLoadStateLocation(t *testing.T, name string) *time.Location {
t.Helper()
location, err := time.LoadLocation(name)
if err != nil {
t.Fatalf("LoadLocation(%q) error = %v", name, err)
}
return location
}
func resolveDailyAt(t *testing.T, value string) report.Resolved {
t.Helper()
return resolveDailyForDateAt(t, value, value)

34
internal/state/store.go
View File

@@ -17,6 +17,7 @@ type Store interface {
SaveDataPackage(context.Context, report.Resolved, promptinput.Package) (string, error)
SavePreflight(context.Context, report.Resolved, PreflightArtifact) (string, error)
SaveDistributorNotification(context.Context, report.Resolved, DistributorNotificationArtifact) (string, error)
SaveBatchDistributorNotification(context.Context, BatchDistributorNotificationRef, BatchDistributorNotificationArtifact) (string, error)
SaveGeneratedTextRaw(context.Context, report.Resolved, []byte) (string, error)
SaveGeneratedTextResult(context.Context, report.Resolved, any) (string, error)
SaveGeneratedText(context.Context, report.Resolved, []byte) (string, error)
@@ -45,6 +46,14 @@ type PreflightArtifact struct {
}
const DistributorNotificationSchemaVersion = "weatherreporter.distributor_notification.v1"
const BatchDistributorNotificationSchemaVersion = "weatherreporter.batch_distributor_notification.v1"
type BatchDistributorNotificationRef struct {
Batch string
BatchRunID string
StartedAt time.Time
Location *time.Location
}
type DistributorNotificationArtifact struct {
SchemaVersion string `json:"schemaVersion"`
@@ -80,3 +89,28 @@ type DistributorRunStatus struct {
Report json.RawMessage `json:"report,omitempty"`
Error string `json:"error,omitempty"`
}
type BatchDistributorNotificationArtifact struct {
SchemaVersion string `json:"schemaVersion"`
Batch string `json:"batch"`
BatchRunID string `json:"batchRunId"`
AttemptedAt time.Time `json:"attemptedAt"`
Endpoint string `json:"endpoint"`
PipelineID string `json:"pipelineId,omitempty"`
BundleID string `json:"bundleId,omitempty"`
IdempotencyKey string `json:"idempotencyKey,omitempty"`
BundleCreated time.Time `json:"bundleCreated,omitempty"`
Reports []BatchDistributorNotificationReportArtifact `json:"includedReports,omitempty"`
Status string `json:"status"`
Upload *DistributorUploadResult `json:"upload,omitempty"`
RunStatus *DistributorRunStatus `json:"runStatus,omitempty"`
StatusError string `json:"statusError,omitempty"`
Error string `json:"error,omitempty"`
}
type BatchDistributorNotificationReportArtifact struct {
ReportID report.ID `json:"reportId"`
RunID string `json:"runId"`
SourcePath string `json:"sourcePath"`
BundlePaths []string `json:"bundlePaths"`
}