317 lines
11 KiB
Go
317 lines
11 KiB
Go
package state
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/config"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/fileutil"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/module"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/promptinput"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/report"
|
|
)
|
|
|
|
type FilesystemStore struct {
|
|
root string
|
|
snapshotsDir string
|
|
reportsDir string
|
|
dataPackagesDir string
|
|
preflightDir string
|
|
notificationsDir string
|
|
}
|
|
|
|
type ArtifactPaths struct {
|
|
ModuleSnapshot string `json:"moduleSnapshot"`
|
|
Metadata string `json:"metadata"`
|
|
DataPackage string `json:"dataPackage"`
|
|
Preparation string `json:"preparation,omitempty"`
|
|
Execution string `json:"execution,omitempty"`
|
|
Notification string `json:"notification,omitempty"`
|
|
RenderedReport string `json:"renderedReport,omitempty"`
|
|
GeneratedTextRaw string `json:"generatedTextRaw,omitempty"`
|
|
GeneratedText string `json:"generatedText,omitempty"`
|
|
RenderContext string `json:"renderContext,omitempty"`
|
|
}
|
|
|
|
func NewFilesystemStore(cfg config.WorkspaceConfig) (*FilesystemStore, error) {
|
|
if cfg.Root == "" {
|
|
return nil, fmt.Errorf("workspace root is required")
|
|
}
|
|
for name, value := range map[string]string{
|
|
"snapshots_dir": cfg.SnapshotsDir,
|
|
"reports_dir": cfg.ReportsDir,
|
|
"data_packages_dir": cfg.DataPackagesDir,
|
|
"preflight_dir": cfg.PreflightDir,
|
|
"notifications_dir": cfg.NotificationsDir,
|
|
} {
|
|
if err := validateRelativeDir(name, value); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return &FilesystemStore{
|
|
root: filepath.Clean(cfg.Root),
|
|
snapshotsDir: filepath.Clean(cfg.SnapshotsDir),
|
|
reportsDir: filepath.Clean(cfg.ReportsDir),
|
|
dataPackagesDir: filepath.Clean(cfg.DataPackagesDir),
|
|
preflightDir: filepath.Clean(cfg.PreflightDir),
|
|
notificationsDir: filepath.Clean(cfg.NotificationsDir),
|
|
}, nil
|
|
}
|
|
|
|
func (s *FilesystemStore) Paths(resolved report.Resolved) (ArtifactPaths, error) {
|
|
if s == nil {
|
|
return ArtifactPaths{}, fmt.Errorf("state store is required")
|
|
}
|
|
metadata := resolved.Metadata()
|
|
if err := validatePathSegment("run id", metadata.RunID); err != nil {
|
|
return ArtifactPaths{}, err
|
|
}
|
|
group := resolved.Definition.ArtifactGroup
|
|
if group == "" {
|
|
return ArtifactPaths{}, fmt.Errorf("report %q has no artifact group", resolved.Definition.ID)
|
|
}
|
|
validDate := resolved.ValidPeriod.Start.Format("2006-01-02")
|
|
return ArtifactPaths{
|
|
ModuleSnapshot: s.join(s.snapshotsDir, group, validDate, "modules."+metadata.RunID+".json"),
|
|
Metadata: s.join(s.snapshotsDir, group, validDate, "metadata."+metadata.RunID+".json"),
|
|
DataPackage: s.join(s.dataPackagesDir, group, validDate, "data_package."+metadata.RunID+".yaml"),
|
|
Preparation: s.join(s.preflightDir, group, validDate, "prompt_preparation."+metadata.RunID+".json"),
|
|
Execution: s.join(s.snapshotsDir, group, validDate, "prompt_execution."+metadata.RunID+".json"),
|
|
Notification: s.join(s.notificationsDir, group, validDate, "distributor."+metadata.RunID+".json"),
|
|
RenderedReport: s.join(s.reportsDir, group, validDate, "report."+metadata.RunID+".md"),
|
|
GeneratedTextRaw: s.join(s.snapshotsDir, group, validDate, "generated_text_raw."+metadata.RunID+".json"),
|
|
GeneratedText: s.join(s.snapshotsDir, group, validDate, "generated_text."+metadata.RunID+".json"),
|
|
RenderContext: s.join(s.snapshotsDir, group, validDate, "render_context."+metadata.RunID+".json"),
|
|
}, nil
|
|
}
|
|
|
|
func (s *FilesystemStore) SaveModuleSnapshot(_ context.Context, resolved report.Resolved, snapshot module.Snapshot) (string, error) {
|
|
if err := snapshot.Validate(); err != nil {
|
|
return "", err
|
|
}
|
|
return s.saveResolvedJSON(resolved, func(paths ArtifactPaths) string {
|
|
return paths.ModuleSnapshot
|
|
}, snapshot)
|
|
}
|
|
|
|
func (s *FilesystemStore) SaveDataPackage(ctx context.Context, resolved report.Resolved, pkg promptinput.Package) (string, error) {
|
|
data, err := promptinput.MarshalYAML(pkg)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return s.SaveDataPackageBytes(ctx, resolved, data)
|
|
}
|
|
|
|
func (s *FilesystemStore) SaveDataPackageBytes(_ context.Context, resolved report.Resolved, data []byte) (string, error) {
|
|
paths, err := s.Paths(resolved)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if err := fileutil.WriteFileAtomic(paths.DataPackage, data); err != nil {
|
|
return "", err
|
|
}
|
|
return paths.DataPackage, nil
|
|
}
|
|
|
|
func (s *FilesystemStore) SavePromptPreparation(_ context.Context, resolved report.Resolved, artifact PromptPreparationArtifact) (string, error) {
|
|
if artifact.SchemaVersion == "" {
|
|
artifact.SchemaVersion = PromptPreparationSchemaVersion
|
|
}
|
|
if err := artifact.Validate(); err != nil {
|
|
return "", err
|
|
}
|
|
return s.saveResolvedJSON(resolved, func(paths ArtifactPaths) string {
|
|
return paths.Preparation
|
|
}, artifact)
|
|
}
|
|
|
|
func (s *FilesystemStore) SavePromptExecution(_ context.Context, resolved report.Resolved, artifact PromptExecutionArtifact) (string, error) {
|
|
if artifact.SchemaVersion == "" {
|
|
artifact.SchemaVersion = PromptExecutionSchemaVersion
|
|
}
|
|
if err := artifact.Validate(); err != nil {
|
|
return "", err
|
|
}
|
|
return s.saveResolvedJSON(resolved, func(paths ArtifactPaths) string {
|
|
return paths.Execution
|
|
}, artifact)
|
|
}
|
|
|
|
func (s *FilesystemStore) SaveDistributorNotification(_ context.Context, resolved report.Resolved, artifact DistributorNotificationArtifact) (string, error) {
|
|
if artifact.SchemaVersion == "" {
|
|
artifact.SchemaVersion = DistributorNotificationSchemaVersion
|
|
}
|
|
return s.saveResolvedJSON(resolved, func(paths ArtifactPaths) string {
|
|
return paths.Notification
|
|
}, 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, "distributor."+ref.BatchRunID+".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
|
|
}, data)
|
|
}
|
|
|
|
func (s *FilesystemStore) SaveGeneratedText(_ context.Context, resolved report.Resolved, data []byte) (string, error) {
|
|
return s.saveResolvedBytes(resolved, func(paths ArtifactPaths) string {
|
|
return paths.GeneratedText
|
|
}, data)
|
|
}
|
|
|
|
func (s *FilesystemStore) SaveRenderContext(_ context.Context, resolved report.Resolved, value any) (string, error) {
|
|
return s.saveResolvedJSON(resolved, func(paths ArtifactPaths) string {
|
|
return paths.RenderContext
|
|
}, value)
|
|
}
|
|
|
|
func (s *FilesystemStore) saveResolvedJSON(resolved report.Resolved, selectPath func(ArtifactPaths) string, value any) (string, error) {
|
|
paths, err := s.Paths(resolved)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
path := selectPath(paths)
|
|
if err := fileutil.WriteJSONAtomic(path, value); err != nil {
|
|
return "", err
|
|
}
|
|
return path, nil
|
|
}
|
|
|
|
func (s *FilesystemStore) saveResolvedBytes(resolved report.Resolved, selectPath func(ArtifactPaths) string, data []byte) (string, error) {
|
|
paths, err := s.Paths(resolved)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
path := selectPath(paths)
|
|
if err := fileutil.WriteFileAtomic(path, data); err != nil {
|
|
return "", err
|
|
}
|
|
return path, nil
|
|
}
|
|
|
|
func (s *FilesystemStore) PrepareRenderedReport(_ context.Context, resolved report.Resolved) (string, error) {
|
|
paths, err := s.Paths(resolved)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if err := os.MkdirAll(filepath.Dir(paths.RenderedReport), 0o755); err != nil {
|
|
return "", fmt.Errorf("create rendered report directory %q: %w", filepath.Dir(paths.RenderedReport), err)
|
|
}
|
|
return paths.RenderedReport, nil
|
|
}
|
|
|
|
func (s *FilesystemStore) SaveMetadata(_ context.Context, metadata Metadata) (string, error) {
|
|
if metadata.SchemaVersion != MetadataSchemaVersion {
|
|
return "", fmt.Errorf("new metadata must use schema version %q", MetadataSchemaVersion)
|
|
}
|
|
if err := metadata.Validate(); err != nil {
|
|
return "", err
|
|
}
|
|
if err := s.validateManagedPath("metadata path", metadata.MetadataPath); err != nil {
|
|
return "", err
|
|
}
|
|
if err := fileutil.WriteJSONAtomic(metadata.MetadataPath, metadata); err != nil {
|
|
return "", err
|
|
}
|
|
return metadata.MetadataPath, nil
|
|
}
|
|
|
|
func (s *FilesystemStore) join(parts ...string) string {
|
|
all := append([]string{s.root}, parts...)
|
|
return filepath.Join(all...)
|
|
}
|
|
|
|
func (s *FilesystemStore) validateManagedPath(name, path string) error {
|
|
if s == nil {
|
|
return fmt.Errorf("state store is required")
|
|
}
|
|
root, err := filepath.Abs(s.root)
|
|
if err != nil {
|
|
return fmt.Errorf("resolve workspace root: %w", err)
|
|
}
|
|
target, err := filepath.Abs(path)
|
|
if err != nil {
|
|
return fmt.Errorf("resolve %s: %w", name, err)
|
|
}
|
|
relative, err := filepath.Rel(root, target)
|
|
if err != nil {
|
|
return fmt.Errorf("resolve %s relative to workspace root: %w", name, err)
|
|
}
|
|
if relative == "." || relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
|
|
return fmt.Errorf("%s must stay within workspace root", name)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateRelativeDir(name string, value string) error {
|
|
if value == "" {
|
|
return fmt.Errorf("%s is required", name)
|
|
}
|
|
if filepath.IsAbs(value) {
|
|
return fmt.Errorf("%s must be relative to workspace root", name)
|
|
}
|
|
cleaned := filepath.Clean(value)
|
|
if cleaned == "." || cleaned == ".." || strings.HasPrefix(cleaned, ".."+string(filepath.Separator)) {
|
|
return fmt.Errorf("%s must stay within workspace root", name)
|
|
}
|
|
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
|
|
}
|