From b7db3993fb3c7d0a962e67cfb54564f68ace57bc Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 8 Jun 2026 18:24:12 +0000 Subject: [PATCH] Add reconciliation state foundation --- internal/config/config.go | 41 ++++--- internal/config/defaults.go | 8 ++ internal/config/load_test.go | 32 ++++++ internal/config/validate.go | 8 ++ internal/config/validate_test.go | 34 ++++++ internal/publish/execute.go | 16 ++- internal/publish/output.go | 20 ++++ internal/state/compare_test.go | 18 ++- internal/state/distributor.go | 174 +++++++++++++++++++++++++---- internal/state/distributor_test.go | 157 +++++++++++++++++++++++--- internal/state/outputs.go | 82 ++++++++++++++ internal/state/validate.go | 19 ++++ internal/testutil/fixtures.go | 14 ++- 13 files changed, 557 insertions(+), 66 deletions(-) create mode 100644 internal/state/outputs.go diff --git a/internal/config/config.go b/internal/config/config.go index 047af4a..f754a5d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -38,24 +38,25 @@ type Pipeline struct { } type Destination struct { - ID string `yaml:"id"` - Backend string `yaml:"backend"` - Host string `yaml:"host"` - User string `yaml:"user"` - Port int `yaml:"port"` - Path string `yaml:"path"` - Endpoint string `yaml:"endpoint"` - Bucket string `yaml:"bucket"` - Prefix string `yaml:"prefix"` - Region string `yaml:"region"` - ForcePath *bool `yaml:"force_path_style"` - Creds Credentials `yaml:"credentials"` - SSH SSH `yaml:",inline"` - Publish *PublishPolicy `yaml:"publish"` - Transform Transform `yaml:"transform"` - PathMap PathMapping `yaml:"path_mapping"` - Links *Links `yaml:"links"` - Transfer TransferPolicy `yaml:"transfer"` + ID string `yaml:"id"` + Backend string `yaml:"backend"` + Host string `yaml:"host"` + User string `yaml:"user"` + Port int `yaml:"port"` + Path string `yaml:"path"` + Endpoint string `yaml:"endpoint"` + Bucket string `yaml:"bucket"` + Prefix string `yaml:"prefix"` + Region string `yaml:"region"` + ForcePath *bool `yaml:"force_path_style"` + Creds Credentials `yaml:"credentials"` + SSH SSH `yaml:",inline"` + Publish *PublishPolicy `yaml:"publish"` + Transform Transform `yaml:"transform"` + PathMap PathMapping `yaml:"path_mapping"` + Links *Links `yaml:"links"` + Reconciliation ReconciliationPolicy `yaml:"reconciliation"` + Transfer TransferPolicy `yaml:"transfer"` } type Backend struct { @@ -118,6 +119,10 @@ type Links struct { Primary string `yaml:"primary"` } +type ReconciliationPolicy struct { + Mode string `yaml:"mode"` +} + type TransferPolicy struct { OnDestinationSame string `yaml:"on_destination_same"` OnDestinationOlder string `yaml:"on_destination_older"` diff --git a/internal/config/defaults.go b/internal/config/defaults.go index 00dcf4f..9b281ce 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -42,6 +42,11 @@ const ( LinkPrimarySource = "source" ) +const ( + ReconciliationModeReplace = "replace" + ReconciliationModeMerge = "merge" +) + const DefaultS3Region = "us-east-1" const ( @@ -79,6 +84,9 @@ func ApplyDefaults(cfg *Config) { if destination.Links != nil && destination.Links.Primary == "" { destination.Links.Primary = LinkPrimaryAuto } + if destination.Reconciliation.Mode == "" { + destination.Reconciliation.Mode = ReconciliationModeReplace + } if destination.Transfer.OnDestinationSame == "" { destination.Transfer.OnDestinationSame = TransferActionSkip } diff --git a/internal/config/load_test.go b/internal/config/load_test.go index 674eddc..727a113 100644 --- a/internal/config/load_test.go +++ b/internal/config/load_test.go @@ -33,6 +33,9 @@ pipelines: if got, want := destination.Transfer.OnDestinationOlder, TransferActionReplace; got != want { t.Fatalf("transfer default = %q, want %q", got, want) } + if got, want := destination.Reconciliation.Mode, ReconciliationModeReplace; got != want { + t.Fatalf("reconciliation mode default = %q, want %q", got, want) + } if cfg.Secrets.Directory != "" { t.Fatalf("secrets.directory = %q, want empty", cfg.Secrets.Directory) } @@ -164,6 +167,35 @@ pipelines: } } +func TestLoadFileAcceptsExplicitReconciliationModes(t *testing.T) { + cfg := loadConfig(t, ` +pipelines: + - id: reports + source: + backend: local + path: /source + destinations: + - id: archive + backend: local + path: /archive + reconciliation: + mode: replace + - id: web + backend: local + path: /web + reconciliation: + mode: merge +`) + + destinations := cfg.Pipelines[0].Destinations + if got, want := destinations[0].Reconciliation.Mode, ReconciliationModeReplace; got != want { + t.Fatalf("archive reconciliation mode = %q, want %q", got, want) + } + if got, want := destinations[1].Reconciliation.Mode, ReconciliationModeMerge; got != want { + t.Fatalf("web reconciliation mode = %q, want %q", got, want) + } +} + func TestLoadFileAcceptsFixedPathMapping(t *testing.T) { cfg := loadConfig(t, ` pipelines: diff --git a/internal/config/validate.go b/internal/config/validate.go index 52d912e..260d18b 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -72,6 +72,7 @@ func Validate(cfg Config) error { errs = validatePublishTransformPolicy(errs, destinationContext, destination.Publish, destination.Transform) errs = validatePathMapping(errs, destinationContext+".path_mapping", destination.PathMap) errs = validateLinks(errs, destinationContext+".links", destination.Links) + errs = validateReconciliationPolicy(errs, destinationContext+".reconciliation", destination.Reconciliation) errs = validateTransferPolicy(errs, destinationContext+".transfer", destination.Transfer) } } @@ -314,6 +315,13 @@ func validateLinks(errs ValidationErrors, context string, links *Links) Validati return errs } +func validateReconciliationPolicy(errs ValidationErrors, context string, policy ReconciliationPolicy) ValidationErrors { + if policy.Mode != ReconciliationModeReplace && policy.Mode != ReconciliationModeMerge { + errs = append(errs, context+".mode must be "+ReconciliationModeReplace+" or "+ReconciliationModeMerge) + } + return errs +} + func validateTransferPolicy(errs ValidationErrors, context string, policy TransferPolicy) ValidationErrors { if policy.OnDestinationSame != TransferActionSkip && policy.OnDestinationSame != TransferActionFail { errs = append(errs, context+".on_destination_same must be skip or fail") diff --git a/internal/config/validate_test.go b/internal/config/validate_test.go index a0e7b82..337aa9d 100644 --- a/internal/config/validate_test.go +++ b/internal/config/validate_test.go @@ -107,6 +107,40 @@ func TestValidatePathMapping(t *testing.T) { } } +func TestValidateReconciliationPolicy(t *testing.T) { + tests := []struct { + name string + mode string + wantErr bool + }{ + {name: "replace", mode: ReconciliationModeReplace}, + {name: "merge", mode: ReconciliationModeMerge}, + {name: "invalid", mode: "append", wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := Config{Pipelines: []Pipeline{{ + ID: "reports", + Source: Backend{Backend: BackendLocal, Path: "/source"}, + Destinations: []Destination{{ + ID: "archive", + Backend: BackendLocal, + Path: "/destination", + Reconciliation: ReconciliationPolicy{Mode: tt.mode}, + }}, + }}} + ApplyDefaults(&cfg) + err := Validate(cfg) + if tt.wantErr && err == nil { + t.Fatal("Validate() error = nil, want error") + } + if !tt.wantErr && err != nil { + t.Fatalf("Validate() error = %v", err) + } + }) + } +} + func TestValidateLinks(t *testing.T) { tests := []struct { name string diff --git a/internal/publish/execute.go b/internal/publish/execute.go index a052bc4..21403fa 100644 --- a/internal/publish/execute.go +++ b/internal/publish/execute.go @@ -6,6 +6,7 @@ import ( "fmt" "time" + "gitea.maximumdirect.net/eric/distributor/internal/config" "gitea.maximumdirect.net/eric/distributor/internal/state" "gitea.maximumdirect.net/eric/distributor/internal/storage" ) @@ -69,14 +70,25 @@ func Execute(ctx context.Context, req Request, plan Plan) error { writtenOutputs = append(writtenOutputs, output) } + now := time.Now().UTC() + createdAt := now + existingOutputs := []state.OutputFile(nil) + if plan.ExistingState != nil { + createdAt = plan.ExistingState.CreatedAt + existingOutputs = plan.ExistingState.Outputs + } destinationState := state.DistributorState{ SchemaVersion: state.SchemaVersion, DistributorVersion: req.DistributorVersion, PipelineID: req.PipelineID, DestinationID: req.DestinationID, - PublishedAt: time.Now().UTC(), + PublishedAt: now, + CreatedAt: createdAt, + UpdatedAt: now, + State: state.StatePolicy{Mode: state.StateModeSingleOwner}, + Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace}, Source: state.SourceState{Manifest: req.SourceBundle.Manifest}, - Outputs: StateOutputFiles(plan.Outputs), + Outputs: state.ProjectOutputs(StateOutputProjections(plan.Outputs), existingOutputs, now), } if plan.PrimaryURL != "" { destinationState.Links = &state.LinkState{PrimaryURL: plan.PrimaryURL} diff --git a/internal/publish/output.go b/internal/publish/output.go index 5059318..b7853a2 100644 --- a/internal/publish/output.go +++ b/internal/publish/output.go @@ -109,6 +109,18 @@ func (o Output) StateOutputFile() state.OutputFile { } } +func (o Output) StateOutputProjection() state.OutputProjection { + return state.OutputProjection{ + Path: o.DestinationPath, + Kind: o.Kind, + SourcePath: o.SourcePath, + Transform: o.Transform, + URL: o.URL, + SHA256: o.SHA256, + Size: o.Size, + } +} + func (o Output) ManagedPath() string { return o.DestinationPath } @@ -121,6 +133,14 @@ func StateOutputFiles(outputs []Output) []state.OutputFile { return files } +func StateOutputProjections(outputs []Output) []state.OutputProjection { + projections := make([]state.OutputProjection, 0, len(outputs)) + for _, output := range outputs { + projections = append(projections, output.StateOutputProjection()) + } + return projections +} + func ManagedOutputPaths(outputs []Output) []string { paths := make([]string, 0, len(outputs)) for _, output := range outputs { diff --git a/internal/state/compare_test.go b/internal/state/compare_test.go index e749a9c..8514002 100644 --- a/internal/state/compare_test.go +++ b/internal/state/compare_test.go @@ -6,6 +6,7 @@ import ( "time" "gitea.maximumdirect.net/eric/distributor/internal/bundle" + "gitea.maximumdirect.net/eric/distributor/internal/config" ) func TestCompareOutcomes(t *testing.T) { @@ -100,18 +101,25 @@ func withState(t *testing.T, source bundle.Manifest, mutate func(*DistributorSta t.Helper() stateManifest := source stateManifest.Files = append([]bundle.ManifestFile(nil), source.Files...) + publishedAt := time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC) state := DistributorState{ - SchemaVersion: SchemaVersion, - PipelineID: "reports", - DestinationID: "archive", - PublishedAt: time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC), - Source: SourceState{Manifest: stateManifest}, + SchemaVersion: SchemaVersion, + PipelineID: "reports", + DestinationID: "archive", + PublishedAt: publishedAt, + CreatedAt: publishedAt, + UpdatedAt: publishedAt, + State: StatePolicy{Mode: StateModeSingleOwner}, + Reconciliation: ReconciliationPolicy{Mode: config.ReconciliationModeReplace}, + Source: SourceState{Manifest: stateManifest}, Outputs: []OutputFile{{ Path: "report.md", Kind: OutputKindSource, SourcePath: "report.md", SHA256: source.Files[0].SHA256, Size: source.Files[0].Size, + CreatedAt: publishedAt, + UpdatedAt: publishedAt, }}, } if mutate != nil { diff --git a/internal/state/distributor.go b/internal/state/distributor.go index 13e9df2..d2e445d 100644 --- a/internal/state/distributor.go +++ b/internal/state/distributor.go @@ -8,9 +8,14 @@ import ( "time" "gitea.maximumdirect.net/eric/distributor/internal/bundle" + "gitea.maximumdirect.net/eric/distributor/internal/config" ) -const SchemaVersion = 1 +const ( + SchemaVersion = 2 + legacySchemaVersion = 1 + StateModeSingleOwner = "single_owner" +) type DistributorState struct { SchemaVersion int @@ -18,11 +23,23 @@ type DistributorState struct { PipelineID string DestinationID string PublishedAt time.Time + CreatedAt time.Time + UpdatedAt time.Time + State StatePolicy + Reconciliation ReconciliationPolicy Source SourceState Links *LinkState Outputs []OutputFile } +type StatePolicy struct { + Mode string +} + +type ReconciliationPolicy struct { + Mode string +} + type SourceState struct { Manifest bundle.Manifest } @@ -39,17 +56,31 @@ type OutputFile struct { URL string SHA256 string Size int64 + CreatedAt time.Time + UpdatedAt time.Time } type rawDistributorState struct { - SchemaVersion *int `json:"schema_version"` - DistributorVersion string `json:"distributor_version"` - PipelineID *string `json:"pipeline_id"` - DestinationID *string `json:"destination_id"` - PublishedAt *string `json:"published_at"` - Source *rawSourceState `json:"source"` - Links *rawLinkState `json:"links"` - Outputs []rawOutputFile `json:"outputs"` + SchemaVersion *int `json:"schema_version"` + DistributorVersion string `json:"distributor_version"` + PipelineID *string `json:"pipeline_id"` + DestinationID *string `json:"destination_id"` + PublishedAt *string `json:"published_at"` + CreatedAt *string `json:"created_at"` + UpdatedAt *string `json:"updated_at"` + State *rawStatePolicy `json:"state"` + Reconciliation *rawReconciliationPolicy `json:"reconciliation"` + Source *rawSourceState `json:"source"` + Links *rawLinkState `json:"links"` + Outputs []rawOutputFile `json:"outputs"` +} + +type rawStatePolicy struct { + Mode string `json:"mode"` +} + +type rawReconciliationPolicy struct { + Mode string `json:"mode"` } type rawSourceState struct { @@ -68,6 +99,8 @@ type rawOutputFile struct { URL string `json:"url"` SHA256 *string `json:"sha256"` Size *int64 `json:"size"` + CreatedAt *string `json:"created_at"` + UpdatedAt *string `json:"updated_at"` } func Parse(data []byte) (DistributorState, error) { @@ -96,9 +129,10 @@ func parseRaw(raw rawDistributorState) (DistributorState, error) { return DistributorState{}, fmt.Errorf("state schema_version is required") } state.SchemaVersion = *raw.SchemaVersion - if state.SchemaVersion != SchemaVersion { - return DistributorState{}, fmt.Errorf("state schema_version must be %d", SchemaVersion) + if state.SchemaVersion != SchemaVersion && state.SchemaVersion != legacySchemaVersion { + return DistributorState{}, fmt.Errorf("state schema_version must be %d or %d", legacySchemaVersion, SchemaVersion) } + legacy := state.SchemaVersion == legacySchemaVersion state.DistributorVersion = raw.DistributorVersion if raw.PipelineID == nil || *raw.PipelineID == "" { return DistributorState{}, fmt.Errorf("state pipeline_id is required") @@ -116,6 +150,32 @@ func parseRaw(raw rawDistributorState) (DistributorState, error) { return DistributorState{}, fmt.Errorf("state published_at must be RFC3339: %w", err) } state.PublishedAt = publishedAt.UTC() + if legacy { + state.SchemaVersion = SchemaVersion + state.CreatedAt = state.PublishedAt + state.UpdatedAt = state.PublishedAt + state.State.Mode = StateModeSingleOwner + state.Reconciliation.Mode = config.ReconciliationModeReplace + } else { + createdAt, err := parseRequiredTime("state created_at", raw.CreatedAt) + if err != nil { + return DistributorState{}, err + } + updatedAt, err := parseRequiredTime("state updated_at", raw.UpdatedAt) + if err != nil { + return DistributorState{}, err + } + state.CreatedAt = createdAt + state.UpdatedAt = updatedAt + if raw.State == nil || raw.State.Mode == "" { + return DistributorState{}, fmt.Errorf("state state.mode is required") + } + state.State.Mode = raw.State.Mode + if raw.Reconciliation == nil || raw.Reconciliation.Mode == "" { + return DistributorState{}, fmt.Errorf("state reconciliation.mode is required") + } + state.Reconciliation.Mode = raw.Reconciliation.Mode + } if raw.Source == nil || len(raw.Source.Manifest) == 0 { return DistributorState{}, fmt.Errorf("state source.manifest is required") } @@ -130,7 +190,7 @@ func parseRaw(raw rawDistributorState) (DistributorState, error) { if raw.Outputs == nil { return DistributorState{}, fmt.Errorf("state outputs is required") } - outputs, err := parseOutputs(raw.Outputs) + outputs, err := parseOutputs(raw.Outputs, legacy, state.PublishedAt) if err != nil { return DistributorState{}, err } @@ -138,11 +198,22 @@ func parseRaw(raw rawDistributorState) (DistributorState, error) { return state, nil } -func parseOutputs(rawOutputs []rawOutputFile) ([]OutputFile, error) { +func parseRequiredTime(context string, raw *string) (time.Time, error) { + if raw == nil || *raw == "" { + return time.Time{}, fmt.Errorf("%s is required", context) + } + parsed, err := time.Parse(time.RFC3339, *raw) + if err != nil { + return time.Time{}, fmt.Errorf("%s must be RFC3339: %w", context, err) + } + return parsed.UTC(), nil +} + +func parseOutputs(rawOutputs []rawOutputFile, legacy bool, publishedAt time.Time) ([]OutputFile, error) { outputs := make([]OutputFile, 0, len(rawOutputs)) seen := make(map[string]struct{}, len(rawOutputs)) for index, raw := range rawOutputs { - output, err := parseOutput(index, raw) + output, err := parseOutput(index, raw, legacy, publishedAt) if err != nil { return nil, err } @@ -155,7 +226,7 @@ func parseOutputs(rawOutputs []rawOutputFile) ([]OutputFile, error) { return outputs, nil } -func parseOutput(index int, raw rawOutputFile) (OutputFile, error) { +func parseOutput(index int, raw rawOutputFile, legacy bool, publishedAt time.Time) (OutputFile, error) { if raw.Path == nil || *raw.Path == "" { return OutputFile{}, fmt.Errorf("state outputs[%d].path is required", index) } @@ -171,6 +242,19 @@ func parseOutput(index int, raw rawOutputFile) (OutputFile, error) { if raw.Size == nil { return OutputFile{}, fmt.Errorf("state outputs[%d].size is required", index) } + createdAt := publishedAt + updatedAt := publishedAt + if !legacy { + var err error + createdAt, err = parseRequiredTime(fmt.Sprintf("state outputs[%d].created_at", index), raw.CreatedAt) + if err != nil { + return OutputFile{}, err + } + updatedAt, err = parseRequiredTime(fmt.Sprintf("state outputs[%d].updated_at", index), raw.UpdatedAt) + if err != nil { + return OutputFile{}, err + } + } return OutputFile{ Path: *raw.Path, Kind: *raw.Kind, @@ -179,6 +263,8 @@ func parseOutput(index int, raw rawOutputFile) (OutputFile, error) { URL: raw.URL, SHA256: *raw.SHA256, Size: *raw.Size, + CreatedAt: createdAt, + UpdatedAt: updatedAt, }, nil } @@ -186,19 +272,39 @@ func (s DistributorState) PublishedAtString() string { return s.PublishedAt.UTC().Format(time.RFC3339) } +func (s DistributorState) CreatedAtString() string { + return s.CreatedAt.UTC().Format(time.RFC3339) +} + +func (s DistributorState) UpdatedAtString() string { + return s.UpdatedAt.UTC().Format(time.RFC3339) +} + +func (o OutputFile) CreatedAtString() string { + return o.CreatedAt.UTC().Format(time.RFC3339) +} + +func (o OutputFile) UpdatedAtString() string { + return o.UpdatedAt.UTC().Format(time.RFC3339) +} + func (s DistributorState) MarshalJSON() ([]byte, error) { type sourceJSON struct { Manifest bundle.Manifest `json:"manifest"` } type stateJSON struct { - SchemaVersion int `json:"schema_version"` - DistributorVersion string `json:"distributor_version,omitempty"` - PipelineID string `json:"pipeline_id"` - DestinationID string `json:"destination_id"` - PublishedAt string `json:"published_at"` - Source sourceJSON `json:"source"` - Links *LinkState `json:"links,omitempty"` - Outputs []OutputFile `json:"outputs"` + SchemaVersion int `json:"schema_version"` + DistributorVersion string `json:"distributor_version,omitempty"` + PipelineID string `json:"pipeline_id"` + DestinationID string `json:"destination_id"` + PublishedAt string `json:"published_at"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` + State StatePolicy `json:"state"` + Reconciliation ReconciliationPolicy `json:"reconciliation"` + Source sourceJSON `json:"source"` + Links *LinkState `json:"links,omitempty"` + Outputs []OutputFile `json:"outputs"` } return json.Marshal(stateJSON{ SchemaVersion: s.SchemaVersion, @@ -206,12 +312,30 @@ func (s DistributorState) MarshalJSON() ([]byte, error) { PipelineID: s.PipelineID, DestinationID: s.DestinationID, PublishedAt: s.PublishedAtString(), + CreatedAt: s.CreatedAtString(), + UpdatedAt: s.UpdatedAtString(), + State: s.State, + Reconciliation: s.Reconciliation, Source: sourceJSON{Manifest: s.Source.Manifest}, Links: s.Links, Outputs: s.Outputs, }) } +func (p StatePolicy) MarshalJSON() ([]byte, error) { + type policyJSON struct { + Mode string `json:"mode"` + } + return json.Marshal(policyJSON{Mode: p.Mode}) +} + +func (p ReconciliationPolicy) MarshalJSON() ([]byte, error) { + type policyJSON struct { + Mode string `json:"mode"` + } + return json.Marshal(policyJSON{Mode: p.Mode}) +} + func (l LinkState) MarshalJSON() ([]byte, error) { type linkJSON struct { PrimaryURL string `json:"primary_url,omitempty"` @@ -228,6 +352,8 @@ func (o OutputFile) MarshalJSON() ([]byte, error) { URL string `json:"url,omitempty"` SHA256 string `json:"sha256"` Size int64 `json:"size"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` } return json.Marshal(outputJSON{ Path: o.Path, @@ -237,5 +363,7 @@ func (o OutputFile) MarshalJSON() ([]byte, error) { URL: o.URL, SHA256: o.SHA256, Size: o.Size, + CreatedAt: o.CreatedAt.UTC().Format(time.RFC3339), + UpdatedAt: o.UpdatedAt.UTC().Format(time.RFC3339), }) } diff --git a/internal/state/distributor_test.go b/internal/state/distributor_test.go index c5bc7f6..ecc56fd 100644 --- a/internal/state/distributor_test.go +++ b/internal/state/distributor_test.go @@ -8,6 +8,7 @@ import ( "time" "gitea.maximumdirect.net/eric/distributor/internal/bundle" + "gitea.maximumdirect.net/eric/distributor/internal/config" ) func TestParseValidState(t *testing.T) { @@ -24,9 +25,24 @@ func TestParseValidState(t *testing.T) { if got, want := state.PublishedAtString(), "2026-05-30T11:12:00Z"; got != want { t.Fatalf("PublishedAtString() = %q, want %q", got, want) } + if got, want := state.CreatedAtString(), "2026-05-30T11:12:00Z"; got != want { + t.Fatalf("CreatedAtString() = %q, want %q", got, want) + } + if got, want := state.UpdatedAtString(), "2026-05-30T11:12:00Z"; got != want { + t.Fatalf("UpdatedAtString() = %q, want %q", got, want) + } + if got, want := state.State.Mode, StateModeSingleOwner; got != want { + t.Fatalf("state mode = %q, want %q", got, want) + } + if got, want := state.Reconciliation.Mode, config.ReconciliationModeReplace; got != want { + t.Fatalf("reconciliation mode = %q, want %q", got, want) + } if got, want := len(state.Outputs), 1; got != want { t.Fatalf("output count = %d, want %d", got, want) } + if got, want := state.Outputs[0].CreatedAtString(), "2026-05-30T11:12:00Z"; got != want { + t.Fatalf("output CreatedAtString() = %q, want %q", got, want) + } } func TestParseValidStateWithLinks(t *testing.T) { @@ -62,6 +78,10 @@ func TestParseRejectsMissingFields(t *testing.T) { "pipeline_id": `"pipeline_id"`, "destination_id": `"destination_id"`, "published_at": `"published_at"`, + "created_at": `"created_at"`, + "updated_at": `"updated_at"`, + "state": `"state"`, + "reconciliation": `"reconciliation"`, "source": `"source"`, "outputs": `"outputs"`, } @@ -75,9 +95,31 @@ func TestParseRejectsMissingFields(t *testing.T) { } func TestParseRejectsInvalidSchemaVersion(t *testing.T) { - body := strings.Replace(validStateJSON(t), `"schema_version": 1`, `"schema_version": 2`, 1) + body := strings.Replace(validStateJSON(t), `"schema_version": 2`, `"schema_version": 3`, 1) _, err := Parse([]byte(body)) - assertStateErrorContains(t, err, "schema_version must be 1") + assertStateErrorContains(t, err, "schema_version must be 1 or 2") +} + +func TestParseLegacyStateInfersSingleOwnerDefaults(t *testing.T) { + state, err := Parse([]byte(legacyStateJSON(t))) + if err != nil { + t.Fatalf("Parse() error = %v", err) + } + if got, want := state.SchemaVersion, SchemaVersion; got != want { + t.Fatalf("schema version = %d, want normalized %d", got, want) + } + if got, want := state.CreatedAtString(), "2026-05-30T11:12:00Z"; got != want { + t.Fatalf("created_at = %q, want %q", got, want) + } + if got, want := state.State.Mode, StateModeSingleOwner; got != want { + t.Fatalf("state mode = %q, want %q", got, want) + } + if got, want := state.Reconciliation.Mode, config.ReconciliationModeReplace; got != want { + t.Fatalf("reconciliation mode = %q, want %q", got, want) + } + if got, want := state.Outputs[0].UpdatedAtString(), "2026-05-30T11:12:00Z"; got != want { + t.Fatalf("output updated_at = %q, want %q", got, want) + } } func TestParseRejectsInvalidEmbeddedManifest(t *testing.T) { @@ -161,6 +203,12 @@ func TestParseRejectsInvalidOutputMetadata(t *testing.T) { "negative size": func(s *DistributorState) { s.Outputs[0].Size = -1 }, + "missing output created_at": func(s *DistributorState) { + s.Outputs[0].CreatedAt = time.Time{} + }, + "missing output updated_at": func(s *DistributorState) { + s.Outputs[0].UpdatedAt = time.Time{} + }, "invalid output url": func(s *DistributorState) { s.Outputs[0].URL = "file:///tmp/report.md" }, @@ -202,20 +250,33 @@ func TestParseRejectsMalformedPublishedTimestamp(t *testing.T) { assertStateErrorContains(t, err, "published_at must be RFC3339") } +func TestParseRejectsMalformedCreatedTimestamp(t *testing.T) { + body := strings.Replace(validStateJSON(t), `"created_at": "2026-05-30T11:12:00Z"`, `"created_at": "May 30"`, 1) + _, err := Parse([]byte(body)) + assertStateErrorContains(t, err, "created_at must be RFC3339") +} + func TestMarshalNormalizesPublishedAtUTC(t *testing.T) { source := validManifest(t) + publishedAt := time.Date(2026, 5, 30, 13, 12, 0, 0, time.FixedZone("offset", 2*60*60)) state := DistributorState{ - SchemaVersion: SchemaVersion, - PipelineID: "reports", - DestinationID: "archive", - PublishedAt: time.Date(2026, 5, 30, 13, 12, 0, 0, time.FixedZone("offset", 2*60*60)), - Source: SourceState{Manifest: source}, + SchemaVersion: SchemaVersion, + PipelineID: "reports", + DestinationID: "archive", + PublishedAt: publishedAt, + CreatedAt: publishedAt, + UpdatedAt: publishedAt, + State: StatePolicy{Mode: StateModeSingleOwner}, + Reconciliation: ReconciliationPolicy{Mode: config.ReconciliationModeReplace}, + Source: SourceState{Manifest: source}, Outputs: []OutputFile{{ Path: "report.md", Kind: OutputKindSource, SourcePath: "report.md", SHA256: source.Files[0].SHA256, Size: source.Files[0].Size, + CreatedAt: publishedAt, + UpdatedAt: publishedAt, }}, } data, err := json.Marshal(state) @@ -229,13 +290,18 @@ func TestMarshalNormalizesPublishedAtUTC(t *testing.T) { func TestMarshalIncludesLinksWhenPresent(t *testing.T) { source := validManifest(t) + publishedAt := time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC) state := DistributorState{ - SchemaVersion: SchemaVersion, - PipelineID: "reports", - DestinationID: "archive", - PublishedAt: time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC), - Source: SourceState{Manifest: source}, - Links: &LinkState{PrimaryURL: "https://reports.example.com/archive/report.md"}, + SchemaVersion: SchemaVersion, + PipelineID: "reports", + DestinationID: "archive", + PublishedAt: publishedAt, + CreatedAt: publishedAt, + UpdatedAt: publishedAt, + State: StatePolicy{Mode: StateModeSingleOwner}, + Reconciliation: ReconciliationPolicy{Mode: config.ReconciliationModeReplace}, + Source: SourceState{Manifest: source}, + Links: &LinkState{PrimaryURL: "https://reports.example.com/archive/report.md"}, Outputs: []OutputFile{{ Path: "report.md", Kind: OutputKindSource, @@ -243,6 +309,8 @@ func TestMarshalIncludesLinksWhenPresent(t *testing.T) { URL: "https://reports.example.com/archive/report.md", SHA256: source.Files[0].SHA256, Size: source.Files[0].Size, + CreatedAt: publishedAt, + UpdatedAt: publishedAt, }}, } data, err := json.Marshal(state) @@ -257,6 +325,42 @@ func TestMarshalIncludesLinksWhenPresent(t *testing.T) { } } +func TestOutputHelpers(t *testing.T) { + createdAt := time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC) + updatedAt := createdAt.Add(time.Hour) + retained := []OutputFile{{ + Path: "old.md", + Kind: OutputKindSource, + CreatedAt: createdAt, + UpdatedAt: createdAt, + }, { + Path: "report.md", + Kind: OutputKindSource, + CreatedAt: createdAt, + UpdatedAt: createdAt, + }} + projected := ProjectOutputs([]OutputProjection{{ + Path: "report.md", + Kind: OutputKindSource, + }, { + Path: "new.md", + Kind: OutputKindSource, + }}, retained, updatedAt) + if got, ok := FindOutputByPath(projected, "report.md"); !ok || !got.CreatedAt.Equal(createdAt) || !got.UpdatedAt.Equal(updatedAt) { + t.Fatalf("projected report.md = %#v, want preserved created_at and updated updated_at", got) + } + merged, err := MergeOutputFiles(retained, projected) + if err != nil { + t.Fatalf("MergeOutputFiles() error = %v", err) + } + if got, want := len(merged), 3; got != want { + t.Fatalf("merged count = %d, want %d", got, want) + } + if got, want := ManagedOutputPaths(DistributorState{Outputs: merged}), []string{"old.md", "report.md", "new.md"}; strings.Join(got, ",") != strings.Join(want, ",") { + t.Fatalf("managed paths = %#v, want %#v", got, want) + } +} + func validStateJSON(t *testing.T) string { t.Helper() return validStateWithManifestJSON(t, manifestJSON(t)) @@ -265,11 +369,15 @@ func validStateJSON(t *testing.T) string { func validStateWithManifestJSON(t *testing.T, manifest string) string { t.Helper() return `{ - "schema_version": 1, + "schema_version": 2, "distributor_version": "dev", "pipeline_id": "reports", "destination_id": "archive", "published_at": "2026-05-30T11:12:00Z", + "created_at": "2026-05-30T11:12:00Z", + "updated_at": "2026-05-30T11:12:00Z", + "state": {"mode": "single_owner"}, + "reconciliation": {"mode": "replace"}, "source": { "manifest": ` + manifest + ` }, @@ -279,12 +387,31 @@ func validStateWithManifestJSON(t *testing.T, manifest string) string { "kind": "source", "source_path": "report.md", "sha256": "sha256:3640fd37140ee4d2e0e93e78834f232ea67a50e7bc6279203690cc7de1975fa6", - "size": 16 + "size": 16, + "created_at": "2026-05-30T11:12:00Z", + "updated_at": "2026-05-30T11:12:00Z" } ] }` } +func legacyStateJSON(t *testing.T) string { + t.Helper() + return strings.Replace(strings.Replace(strings.Replace(strings.Replace(strings.Replace(strings.Replace(validStateJSON(t), + `"schema_version": 2`, `"schema_version": 1`, 1), + ` "created_at": "2026-05-30T11:12:00Z", +`, "", 1), + ` "updated_at": "2026-05-30T11:12:00Z", +`, "", 1), + ` "state": {"mode": "single_owner"}, +`, "", 1), + ` "reconciliation": {"mode": "replace"}, +`, "", 1), + `, + "created_at": "2026-05-30T11:12:00Z", + "updated_at": "2026-05-30T11:12:00Z"`, "", 1) +} + func manifestJSON(t *testing.T) string { t.Helper() data, err := os.ReadFile("../bundle/testdata/valid_bundle/manifest.json") diff --git a/internal/state/outputs.go b/internal/state/outputs.go new file mode 100644 index 0000000..723daa9 --- /dev/null +++ b/internal/state/outputs.go @@ -0,0 +1,82 @@ +package state + +import ( + "fmt" + "time" +) + +type OutputProjection struct { + Path string + Kind string + SourcePath string + Transform string + URL string + SHA256 string + Size int64 +} + +func FindOutputByPath(outputs []OutputFile, path string) (OutputFile, bool) { + for _, output := range outputs { + if output.Path == path { + return output, true + } + } + return OutputFile{}, false +} + +func MergeOutputFiles(retained, planned []OutputFile) ([]OutputFile, error) { + outputs := make([]OutputFile, 0, len(retained)+len(planned)) + indexByPath := make(map[string]int, len(retained)+len(planned)) + for _, output := range retained { + if _, exists := indexByPath[output.Path]; exists { + return nil, fmt.Errorf("state output path %q is duplicated", output.Path) + } + indexByPath[output.Path] = len(outputs) + outputs = append(outputs, output) + } + seenPlanned := make(map[string]struct{}, len(planned)) + for _, output := range planned { + if _, exists := seenPlanned[output.Path]; exists { + return nil, fmt.Errorf("state output path %q is duplicated", output.Path) + } + seenPlanned[output.Path] = struct{}{} + if index, exists := indexByPath[output.Path]; exists { + outputs[index] = output + continue + } + indexByPath[output.Path] = len(outputs) + outputs = append(outputs, output) + } + return outputs, nil +} + +func ManagedOutputPaths(s DistributorState) []string { + paths := make([]string, 0, len(s.Outputs)) + for _, output := range s.Outputs { + paths = append(paths, output.Path) + } + return paths +} + +func ProjectOutputs(outputs []OutputProjection, existing []OutputFile, now time.Time) []OutputFile { + now = now.UTC() + files := make([]OutputFile, 0, len(outputs)) + for _, output := range outputs { + createdAt := now + if existingOutput, ok := FindOutputByPath(existing, output.Path); ok { + createdAt = existingOutput.CreatedAt + } + files = append(files, OutputFile{ + Path: output.Path, + Kind: output.Kind, + SourcePath: output.SourcePath, + Transform: output.Transform, + URL: output.URL, + SHA256: output.SHA256, + Size: output.Size, + CreatedAt: createdAt, + UpdatedAt: now, + }) + } + return files +} diff --git a/internal/state/validate.go b/internal/state/validate.go index 31a4782..5f8f3d4 100644 --- a/internal/state/validate.go +++ b/internal/state/validate.go @@ -4,6 +4,7 @@ import ( "fmt" "gitea.maximumdirect.net/eric/distributor/internal/bundle" + "gitea.maximumdirect.net/eric/distributor/internal/config" "gitea.maximumdirect.net/eric/distributor/internal/link" "gitea.maximumdirect.net/eric/distributor/internal/storage" ) @@ -26,6 +27,18 @@ func Validate(s DistributorState) error { if s.PublishedAt.IsZero() { return fmt.Errorf("state published_at is required") } + if s.CreatedAt.IsZero() { + return fmt.Errorf("state created_at is required") + } + if s.UpdatedAt.IsZero() { + return fmt.Errorf("state updated_at is required") + } + if s.State.Mode != StateModeSingleOwner { + return fmt.Errorf("state state.mode must be %s", StateModeSingleOwner) + } + if s.Reconciliation.Mode != config.ReconciliationModeReplace && s.Reconciliation.Mode != config.ReconciliationModeMerge { + return fmt.Errorf("state reconciliation.mode must be %s or %s", config.ReconciliationModeReplace, config.ReconciliationModeMerge) + } if err := validateEmbeddedManifest(s.Source.Manifest); err != nil { return fmt.Errorf("state source.manifest: %w", err) } @@ -80,5 +93,11 @@ func validateOutput(index int, output OutputFile) error { if output.Size < 0 { return fmt.Errorf("state outputs[%d].size must be non-negative", index) } + if output.CreatedAt.IsZero() { + return fmt.Errorf("state outputs[%d].created_at is required", index) + } + if output.UpdatedAt.IsZero() { + return fmt.Errorf("state outputs[%d].updated_at is required", index) + } return nil } diff --git a/internal/testutil/fixtures.go b/internal/testutil/fixtures.go index deb6448..c2a1785 100644 --- a/internal/testutil/fixtures.go +++ b/internal/testutil/fixtures.go @@ -11,6 +11,7 @@ import ( "time" "gitea.maximumdirect.net/eric/distributor/internal/bundle" + "gitea.maximumdirect.net/eric/distributor/internal/config" "gitea.maximumdirect.net/eric/distributor/internal/state" "gitea.maximumdirect.net/eric/distributor/internal/storage" "gitea.maximumdirect.net/eric/distributor/internal/storage/fake" @@ -329,14 +330,19 @@ func WriteDestinationState(t testing.TB, root, relative string, manifest bundle. } func DestinationState(manifest bundle.Manifest, opts DestinationStateOptions) state.DistributorState { + publishedAt := defaultPublishedAt(opts.PublishedAt) return state.DistributorState{ SchemaVersion: state.SchemaVersion, DistributorVersion: opts.DistributorVersion, PipelineID: defaultString(opts.PipelineID, "reports"), DestinationID: defaultString(opts.DestinationID, "archive"), - PublishedAt: defaultPublishedAt(opts.PublishedAt), + PublishedAt: publishedAt, + CreatedAt: publishedAt, + UpdatedAt: publishedAt, + State: state.StatePolicy{Mode: state.StateModeSingleOwner}, + Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace}, Source: state.SourceState{Manifest: manifest}, - Outputs: sourceOutputs(manifest), + Outputs: sourceOutputs(manifest, publishedAt), } } @@ -386,7 +392,7 @@ func sourceFiles(opts BundleOptions) []SourceFile { return files } -func sourceOutputs(manifest bundle.Manifest) []state.OutputFile { +func sourceOutputs(manifest bundle.Manifest, publishedAt time.Time) []state.OutputFile { outputs := make([]state.OutputFile, 0, len(manifest.Files)) for _, file := range manifest.Files { outputs = append(outputs, state.OutputFile{ @@ -395,6 +401,8 @@ func sourceOutputs(manifest bundle.Manifest) []state.OutputFile { SourcePath: file.Path, SHA256: file.SHA256, Size: file.Size, + CreatedAt: publishedAt, + UpdatedAt: publishedAt, }) } return outputs