diff --git a/internal/app/prune.go b/internal/app/prune.go index 39693dd..d69e16b 100644 --- a/internal/app/prune.go +++ b/internal/app/prune.go @@ -233,7 +233,7 @@ func removePrunedStateRecords(ctx context.Context, backend storage.Backend, stat } return true, writeRepairedState(ctx, backend, statePath, next) } - return false, fmt.Errorf("destination state document is empty") + return false, unsupportedStateDocumentError(document) } func PlanPrune(document state.StateDocument, policy config.PrunePolicy, options PrunePlanOptions) (PrunePlanReport, error) { @@ -276,7 +276,17 @@ func pruneCandidatesForDocument(document state.StateDocument, scope state.OwnerS if document.SharedRoot != nil { return state.SharedRootPruneCandidates(*document.SharedRoot, scope), nil } - return nil, fmt.Errorf("destination state document is empty") + return nil, unsupportedStateDocumentError(document) +} + +func unsupportedStateDocumentError(document state.StateDocument) error { + if document.SupersededLegacy != nil { + return fmt.Errorf("destination state schema_version %d is superseded legacy state", document.SupersededLegacy.SchemaVersion) + } + if document.Catalog != nil { + return fmt.Errorf("catalog destination state is not supported by this command") + } + return fmt.Errorf("destination state document is empty") } func pruneOlderThan(policy config.PrunePolicy) *time.Duration { diff --git a/internal/app/reconcile_state.go b/internal/app/reconcile_state.go index fbe9ebd..f57d008 100644 --- a/internal/app/reconcile_state.go +++ b/internal/app/reconcile_state.go @@ -161,7 +161,10 @@ func buildReconcileStateReport(ctx context.Context, backend storage.Backend, pip if document.SingleOwner != nil { return reconcileSingleOwnerState(ctx, backend, statePath, *document.SingleOwner, scope, report, options) } - return reconcileSharedRootState(ctx, backend, statePath, *document.SharedRoot, scope, report, options) + if document.SharedRoot != nil { + return reconcileSharedRootState(ctx, backend, statePath, *document.SharedRoot, scope, report, options) + } + return ReconcileStateReport{}, unsupportedStateDocumentError(document) } func reconcileSingleOwnerState(ctx context.Context, backend storage.Backend, statePath string, destinationState state.DistributorState, scope state.OwnerScope, report ReconcileStateReport, options ReconcileStateOptions) (ReconcileStateReport, error) { diff --git a/internal/publish/reconcile.go b/internal/publish/reconcile.go index d29ff7d..ac005ce 100644 --- a/internal/publish/reconcile.go +++ b/internal/publish/reconcile.go @@ -18,7 +18,13 @@ func inspectDestination(ctx context.Context, backend storage.Backend, bundlePath if parseErr != nil { return state.DestinationStatus{StateErr: parseErr}, nil } - return state.DestinationStatus{State: document.SingleOwner, SharedRoot: document.SharedRoot, HasContents: true}, nil + return state.DestinationStatus{ + State: document.SingleOwner, + SharedRoot: document.SharedRoot, + Catalog: document.Catalog, + SupersededLegacy: document.SupersededLegacy, + HasContents: true, + }, nil } if !storage.IsNotFound(err) { return state.DestinationStatus{}, err diff --git a/internal/state/catalog.go b/internal/state/catalog.go new file mode 100644 index 0000000..93d9ccf --- /dev/null +++ b/internal/state/catalog.go @@ -0,0 +1,422 @@ +package state + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "time" + + "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" +) + +type CatalogState struct { + SchemaVersion int + DistributorVersion string + CreatedAt time.Time + UpdatedAt time.Time + State StatePolicy + Outputs []CatalogOutputFile +} + +type CatalogSourceIdentity struct { + ID string + Digest string + Created time.Time +} + +type CatalogOutputFile struct { + Path string + PipelineID string + DestinationID string + Source CatalogSourceIdentity + Kind string + SourcePath string + Transform string + URL string + SHA256 string + Size int64 + CreatedAt time.Time + UpdatedAt time.Time +} + +type rawCatalogState struct { + SchemaVersion *int `json:"schema_version"` + DistributorVersion string `json:"distributor_version"` + CreatedAt *string `json:"created_at"` + UpdatedAt *string `json:"updated_at"` + State *rawStatePolicy `json:"state"` + Outputs []rawCatalogOutput `json:"outputs"` +} + +type rawCatalogOutput struct { + Path *string `json:"path"` + PipelineID *string `json:"pipeline_id"` + DestinationID *string `json:"destination_id"` + Source *rawCatalogSourceIdentity `json:"source"` + Kind *string `json:"kind"` + SourcePath *string `json:"source_path"` + Transform *string `json:"transform"` + URL *string `json:"url"` + SHA256 *string `json:"sha256"` + Size *int64 `json:"size"` + CreatedAt *string `json:"created_at"` + UpdatedAt *string `json:"updated_at"` +} + +type rawCatalogSourceIdentity struct { + ID *string `json:"id"` + Digest *string `json:"digest"` + Created *string `json:"created"` +} + +func ParseCatalog(data []byte) (CatalogState, error) { + decoder := json.NewDecoder(bytes.NewReader(data)) + decoder.DisallowUnknownFields() + var raw rawCatalogState + if err := decoder.Decode(&raw); err != nil { + return CatalogState{}, fmt.Errorf("parse distributor state: %w", err) + } + var extra any + if err := decoder.Decode(&extra); err != io.EOF { + return CatalogState{}, fmt.Errorf("parse distributor state: trailing data") + } + state, err := parseCatalogRaw(raw) + if err != nil { + return CatalogState{}, err + } + if err := ValidateCatalog(state); err != nil { + return CatalogState{}, err + } + return state, nil +} + +func parseCatalogRaw(raw rawCatalogState) (CatalogState, error) { + if raw.SchemaVersion == nil { + return CatalogState{}, fmt.Errorf("state schema_version is required") + } + state := CatalogState{ + SchemaVersion: *raw.SchemaVersion, + DistributorVersion: raw.DistributorVersion, + } + if state.SchemaVersion != CatalogSchemaVersion { + return CatalogState{}, fmt.Errorf("state schema_version must be %d", CatalogSchemaVersion) + } + createdAt, err := parseRequiredTime("state created_at", raw.CreatedAt) + if err != nil { + return CatalogState{}, err + } + updatedAt, err := parseRequiredTime("state updated_at", raw.UpdatedAt) + if err != nil { + return CatalogState{}, err + } + state.CreatedAt = createdAt + state.UpdatedAt = updatedAt + if raw.State == nil || raw.State.Mode == "" { + return CatalogState{}, fmt.Errorf("state state.mode is required") + } + state.State.Mode = raw.State.Mode + if raw.Outputs == nil { + return CatalogState{}, fmt.Errorf("state outputs is required") + } + outputs, err := parseCatalogOutputs(raw.Outputs) + if err != nil { + return CatalogState{}, err + } + state.Outputs = outputs + return state, nil +} + +func parseCatalogOutputs(rawOutputs []rawCatalogOutput) ([]CatalogOutputFile, error) { + outputs := make([]CatalogOutputFile, 0, len(rawOutputs)) + for index, raw := range rawOutputs { + output, err := parseCatalogOutput(index, raw) + if err != nil { + return nil, err + } + outputs = append(outputs, output) + } + return outputs, nil +} + +func parseCatalogOutput(index int, raw rawCatalogOutput) (CatalogOutputFile, error) { + if raw.Path == nil || *raw.Path == "" { + return CatalogOutputFile{}, fmt.Errorf("state outputs[%d].path is required", index) + } + if raw.PipelineID == nil || *raw.PipelineID == "" { + return CatalogOutputFile{}, fmt.Errorf("state outputs[%d].pipeline_id is required", index) + } + if raw.DestinationID == nil || *raw.DestinationID == "" { + return CatalogOutputFile{}, fmt.Errorf("state outputs[%d].destination_id is required", index) + } + source, err := parseCatalogSourceIdentity(index, raw.Source) + if err != nil { + return CatalogOutputFile{}, err + } + if raw.Kind == nil || *raw.Kind == "" { + return CatalogOutputFile{}, fmt.Errorf("state outputs[%d].kind is required", index) + } + switch *raw.Kind { + case OutputKindSource: + if raw.SourcePath != nil { + return CatalogOutputFile{}, fmt.Errorf("state outputs[%d].source_path is only valid for generated output", index) + } + if raw.Transform != nil { + return CatalogOutputFile{}, fmt.Errorf("state outputs[%d].transform is only valid for generated output", index) + } + case OutputKindGenerated: + if raw.SourcePath == nil || *raw.SourcePath == "" { + return CatalogOutputFile{}, fmt.Errorf("state outputs[%d].source_path is required for generated output", index) + } + if raw.Transform == nil || *raw.Transform == "" { + return CatalogOutputFile{}, fmt.Errorf("state outputs[%d].transform is required for generated output", index) + } + } + if raw.SHA256 == nil || *raw.SHA256 == "" { + return CatalogOutputFile{}, fmt.Errorf("state outputs[%d].sha256 is required", index) + } + if raw.Size == nil { + return CatalogOutputFile{}, fmt.Errorf("state outputs[%d].size is required", index) + } + createdAt, err := parseRequiredTime(fmt.Sprintf("state outputs[%d].created_at", index), raw.CreatedAt) + if err != nil { + return CatalogOutputFile{}, err + } + updatedAt, err := parseRequiredTime(fmt.Sprintf("state outputs[%d].updated_at", index), raw.UpdatedAt) + if err != nil { + return CatalogOutputFile{}, err + } + output := CatalogOutputFile{ + Path: *raw.Path, + PipelineID: *raw.PipelineID, + DestinationID: *raw.DestinationID, + Source: source, + Kind: *raw.Kind, + SHA256: *raw.SHA256, + Size: *raw.Size, + CreatedAt: createdAt, + UpdatedAt: updatedAt, + } + if raw.SourcePath != nil { + output.SourcePath = *raw.SourcePath + } + if raw.Transform != nil { + output.Transform = *raw.Transform + } + if raw.URL != nil { + if *raw.URL == "" { + return CatalogOutputFile{}, fmt.Errorf("state outputs[%d].url must not be empty", index) + } + output.URL = *raw.URL + } + return output, nil +} + +func parseCatalogSourceIdentity(index int, raw *rawCatalogSourceIdentity) (CatalogSourceIdentity, error) { + if raw == nil { + return CatalogSourceIdentity{}, fmt.Errorf("state outputs[%d].source is required", index) + } + if raw.ID == nil || *raw.ID == "" { + return CatalogSourceIdentity{}, fmt.Errorf("state outputs[%d].source.id is required", index) + } + if raw.Digest == nil || *raw.Digest == "" { + return CatalogSourceIdentity{}, fmt.Errorf("state outputs[%d].source.digest is required", index) + } + created, err := parseRequiredTime(fmt.Sprintf("state outputs[%d].source.created", index), raw.Created) + if err != nil { + return CatalogSourceIdentity{}, err + } + return CatalogSourceIdentity{ + ID: *raw.ID, + Digest: *raw.Digest, + Created: created, + }, nil +} + +func (s CatalogState) CreatedAtString() string { + return s.CreatedAt.UTC().Format(time.RFC3339) +} + +func (s CatalogState) UpdatedAtString() string { + return s.UpdatedAt.UTC().Format(time.RFC3339) +} + +func (s CatalogSourceIdentity) CreatedString() string { + return s.Created.UTC().Format(time.RFC3339) +} + +func (o CatalogOutputFile) CreatedAtString() string { + return o.CreatedAt.UTC().Format(time.RFC3339) +} + +func (o CatalogOutputFile) UpdatedAtString() string { + return o.UpdatedAt.UTC().Format(time.RFC3339) +} + +func (s CatalogState) MarshalJSON() ([]byte, error) { + type stateJSON struct { + SchemaVersion int `json:"schema_version"` + DistributorVersion string `json:"distributor_version,omitempty"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` + State StatePolicy `json:"state"` + Outputs []CatalogOutputFile `json:"outputs"` + } + return json.Marshal(stateJSON{ + SchemaVersion: s.SchemaVersion, + DistributorVersion: s.DistributorVersion, + CreatedAt: s.CreatedAtString(), + UpdatedAt: s.UpdatedAtString(), + State: s.State, + Outputs: s.Outputs, + }) +} + +func (s CatalogSourceIdentity) MarshalJSON() ([]byte, error) { + type sourceJSON struct { + ID string `json:"id"` + Digest string `json:"digest"` + Created string `json:"created"` + } + return json.Marshal(sourceJSON{ + ID: s.ID, + Digest: s.Digest, + Created: s.CreatedString(), + }) +} + +func (o CatalogOutputFile) MarshalJSON() ([]byte, error) { + type outputJSON struct { + Path string `json:"path"` + PipelineID string `json:"pipeline_id"` + DestinationID string `json:"destination_id"` + Source CatalogSourceIdentity `json:"source"` + Kind string `json:"kind"` + SourcePath string `json:"source_path,omitempty"` + Transform string `json:"transform,omitempty"` + 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, + PipelineID: o.PipelineID, + DestinationID: o.DestinationID, + Source: o.Source, + Kind: o.Kind, + SourcePath: o.SourcePath, + Transform: o.Transform, + URL: o.URL, + SHA256: o.SHA256, + Size: o.Size, + CreatedAt: o.CreatedAtString(), + UpdatedAt: o.UpdatedAtString(), + }) +} + +func ValidateCatalog(s CatalogState) error { + if s.SchemaVersion != CatalogSchemaVersion { + return fmt.Errorf("state schema_version must be %d", CatalogSchemaVersion) + } + 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 != StateModeCatalog { + return fmt.Errorf("state state.mode must be %s", StateModeCatalog) + } + if s.Outputs == nil { + return fmt.Errorf("state outputs is required") + } + seenPaths := make(map[string]struct{}, len(s.Outputs)) + for index, output := range s.Outputs { + if err := validateCatalogOutput(index, output); err != nil { + return err + } + if _, exists := seenPaths[output.Path]; exists { + return fmt.Errorf("state outputs[%d].path duplicates %q", index, output.Path) + } + seenPaths[output.Path] = struct{}{} + } + return nil +} + +func validateCatalogOutput(index int, output CatalogOutputFile) error { + if err := storage.ValidatePath(output.Path); err != nil { + return fmt.Errorf("state outputs[%d].path: %w", index, err) + } + if output.PipelineID == "" { + return fmt.Errorf("state outputs[%d].pipeline_id is required", index) + } + if !config.IsSlugLikeID(output.PipelineID) { + return fmt.Errorf("state outputs[%d].pipeline_id must be a slug-like identifier", index) + } + if output.DestinationID == "" { + return fmt.Errorf("state outputs[%d].destination_id is required", index) + } + if !config.IsSlugLikeID(output.DestinationID) { + return fmt.Errorf("state outputs[%d].destination_id must be a slug-like identifier", index) + } + if err := validateCatalogSourceIdentity(index, output.Source); err != nil { + return err + } + switch output.Kind { + case OutputKindSource: + if output.SourcePath != "" { + return fmt.Errorf("state outputs[%d].source_path is only valid for generated output", index) + } + if output.Transform != "" { + return fmt.Errorf("state outputs[%d].transform is only valid for generated output", index) + } + case OutputKindGenerated: + if output.SourcePath == "" { + return fmt.Errorf("state outputs[%d].source_path is required for generated output", index) + } + if err := storage.ValidatePath(output.SourcePath); err != nil { + return fmt.Errorf("state outputs[%d].source_path: %w", index, err) + } + if output.Transform == "" { + return fmt.Errorf("state outputs[%d].transform is required for generated output", index) + } + default: + return fmt.Errorf("state outputs[%d].kind must be source or generated", index) + } + if output.URL != "" { + if err := link.ValidateHTTPURL(output.URL); err != nil { + return fmt.Errorf("state outputs[%d].url: %w", index, err) + } + } + if err := bundle.ValidateDigest(output.SHA256); err != nil { + return fmt.Errorf("state outputs[%d].sha256: %w", index, err) + } + 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 +} + +func validateCatalogSourceIdentity(index int, source CatalogSourceIdentity) error { + if source.ID == "" { + return fmt.Errorf("state outputs[%d].source.id is required", index) + } + if err := bundle.ValidateDigest(source.Digest); err != nil { + return fmt.Errorf("state outputs[%d].source.digest: %w", index, err) + } + if source.Created.IsZero() { + return fmt.Errorf("state outputs[%d].source.created is required", index) + } + return nil +} diff --git a/internal/state/catalog_test.go b/internal/state/catalog_test.go new file mode 100644 index 0000000..e151cc1 --- /dev/null +++ b/internal/state/catalog_test.go @@ -0,0 +1,361 @@ +package state + +import ( + "encoding/json" + "strings" + "testing" + "time" +) + +func TestParseCatalogState(t *testing.T) { + state, err := ParseCatalog([]byte(validCatalogStateJSON(t))) + if err != nil { + t.Fatalf("ParseCatalog() error = %v", err) + } + if got, want := state.SchemaVersion, CatalogSchemaVersion; got != want { + t.Fatalf("schema version = %d, want %d", got, want) + } + if got, want := state.CreatedAtString(), "2026-06-19T12:00:00Z"; got != want { + t.Fatalf("created_at = %q, want %q", got, want) + } + if got, want := state.UpdatedAtString(), "2026-06-19T12:05:00Z"; got != want { + t.Fatalf("updated_at = %q, want %q", got, want) + } + if got, want := state.State.Mode, StateModeCatalog; got != want { + t.Fatalf("state mode = %q, want %q", got, want) + } + if got, want := len(state.Outputs), 2; got != want { + t.Fatalf("output count = %d, want %d", got, want) + } + sourceOutput := state.Outputs[0] + if sourceOutput.SourcePath != "" || sourceOutput.Transform != "" { + t.Fatalf("source output source_path=%q transform=%q, want omitted", sourceOutput.SourcePath, sourceOutput.Transform) + } + generatedOutput := state.Outputs[1] + if generatedOutput.SourcePath != "report.md" || generatedOutput.Transform != "markdown_to_html" { + t.Fatalf("generated output source_path=%q transform=%q", generatedOutput.SourcePath, generatedOutput.Transform) + } + if got, want := generatedOutput.Source.CreatedString(), "2026-05-30T11:10:00Z"; got != want { + t.Fatalf("source created = %q, want %q", got, want) + } +} + +func TestCatalogMarshalIsDeterministic(t *testing.T) { + data, err := json.Marshal(validCatalogState(t)) + if err != nil { + t.Fatalf("Marshal() error = %v", err) + } + want := `{"schema_version":4,"distributor_version":"dev","created_at":"2026-06-19T12:00:00Z","updated_at":"2026-06-19T12:05:00Z","state":{"mode":"catalog"},"outputs":[{"path":"report.md","pipeline_id":"reports","destination_id":"archive","source":{"id":"weather.daily.brentwood.2026-05-30","digest":"sha256:099b205780d2b050024868399961b05731729a548d5d6329c7b06a6740dd75fe","created":"2026-05-30T11:10:00Z"},"kind":"source","sha256":"sha256:3640fd37140ee4d2e0e93e78834f232ea67a50e7bc6279203690cc7de1975fa6","size":16,"created_at":"2026-06-19T12:00:00Z","updated_at":"2026-06-19T12:05:00Z"},{"path":"report.html","pipeline_id":"reports","destination_id":"html","source":{"id":"weather.daily.brentwood.2026-05-30","digest":"sha256:099b205780d2b050024868399961b05731729a548d5d6329c7b06a6740dd75fe","created":"2026-05-30T11:10:00Z"},"kind":"generated","source_path":"report.md","transform":"markdown_to_html","url":"https://reports.example.com/report.html","sha256":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","size":128,"created_at":"2026-06-19T12:00:00Z","updated_at":"2026-06-19T12:05:00Z"}]}` + if string(data) != want { + t.Fatalf("json = %s, want %s", data, want) + } +} + +func TestParseDocumentHandlesCatalogAndSupersededLegacy(t *testing.T) { + catalog, err := ParseDocument([]byte(validCatalogStateJSON(t))) + if err != nil { + t.Fatalf("ParseDocument(catalog) error = %v", err) + } + if catalog.Catalog == nil || catalog.SingleOwner != nil || catalog.SharedRoot != nil || catalog.SupersededLegacy != nil { + t.Fatalf("catalog document = %#v", catalog) + } + + tests := map[string]string{ + "schema 1": legacyStateJSON(t), + "schema 2": validStateJSON(t), + "schema 3": validSharedRootStateJSON(t), + } + for name, body := range tests { + t.Run(name, func(t *testing.T) { + document, err := ParseDocument([]byte(body)) + if err != nil { + t.Fatalf("ParseDocument() error = %v", err) + } + if document.SupersededLegacy == nil || document.Catalog != nil || document.SingleOwner != nil || document.SharedRoot != nil { + t.Fatalf("document = %#v, want superseded legacy only", document) + } + if document.SupersededLegacy.SchemaVersion < legacySchemaVersion || document.SupersededLegacy.SchemaVersion >= CatalogSchemaVersion { + t.Fatalf("legacy schema version = %d, want 1 through 3", document.SupersededLegacy.SchemaVersion) + } + }) + } +} + +func TestParseDocumentRejectsUnsupportedFutureSchema(t *testing.T) { + body := strings.Replace(validCatalogStateJSON(t), `"schema_version": 4`, `"schema_version": 5`, 1) + _, err := ParseDocument([]byte(body)) + assertStateErrorContains(t, err, "schema_version 5 is unsupported") +} + +func TestParseDocumentRejectsTrailingData(t *testing.T) { + _, err := ParseDocument([]byte(validCatalogStateJSON(t) + "\n{}")) + assertStateErrorContains(t, err, "trailing data") +} + +func TestParseCatalogRejectsMissingFields(t *testing.T) { + tests := map[string]func(map[string]any){ + "schema_version": func(document map[string]any) { + delete(document, "schema_version") + }, + "created_at": func(document map[string]any) { + delete(document, "created_at") + }, + "updated_at": func(document map[string]any) { + delete(document, "updated_at") + }, + "state": func(document map[string]any) { + delete(document, "state") + }, + "outputs": func(document map[string]any) { + delete(document, "outputs") + }, + "path": func(document map[string]any) { + delete(firstCatalogOutput(document), "path") + }, + "pipeline_id": func(document map[string]any) { + delete(firstCatalogOutput(document), "pipeline_id") + }, + "destination_id": func(document map[string]any) { + delete(firstCatalogOutput(document), "destination_id") + }, + "source": func(document map[string]any) { + delete(firstCatalogOutput(document), "source") + }, + "source id": func(document map[string]any) { + delete(firstCatalogSource(document), "id") + }, + "source digest": func(document map[string]any) { + delete(firstCatalogSource(document), "digest") + }, + "source created": func(document map[string]any) { + delete(firstCatalogSource(document), "created") + }, + "kind": func(document map[string]any) { + delete(firstCatalogOutput(document), "kind") + }, + "sha256": func(document map[string]any) { + delete(firstCatalogOutput(document), "sha256") + }, + "size": func(document map[string]any) { + delete(firstCatalogOutput(document), "size") + }, + } + for name, mutate := range tests { + t.Run(name, func(t *testing.T) { + document := catalogStateObject(t) + mutate(document) + _, err := ParseCatalog(mustMarshalCatalogObject(t, document)) + assertStateErrorContains(t, err, "required") + }) + } +} + +func TestParseCatalogRejectsMalformedTimestamps(t *testing.T) { + tests := map[string]func(string) string{ + "created_at": func(body string) string { + return strings.Replace(body, `"created_at": "2026-06-19T12:00:00Z"`, `"created_at": "June 19"`, 1) + }, + "updated_at": func(body string) string { + return strings.Replace(body, `"updated_at": "2026-06-19T12:05:00Z"`, `"updated_at": "June 19"`, 1) + }, + "source created": func(body string) string { + return strings.Replace(body, `"created": "2026-05-30T11:10:00Z"`, `"created": "May 30"`, 1) + }, + "output created_at": func(body string) string { + return strings.Replace(body, ` "created_at": "2026-06-19T12:00:00Z"`, ` "created_at": "June 19"`, 1) + }, + "output updated_at": func(body string) string { + return strings.Replace(body, ` "updated_at": "2026-06-19T12:05:00Z"`, ` "updated_at": "June 19"`, 1) + }, + } + for name, mutate := range tests { + t.Run(name, func(t *testing.T) { + _, err := ParseCatalog([]byte(mutate(validCatalogStateJSON(t)))) + assertStateErrorContains(t, err, "RFC3339") + }) + } +} + +func TestValidateCatalogRejectsInvalidOutputRecords(t *testing.T) { + tests := map[string]func(*CatalogState){ + "duplicate path": func(s *CatalogState) { + s.Outputs[1].Path = s.Outputs[0].Path + }, + "invalid output path": func(s *CatalogState) { + s.Outputs[0].Path = "../report.md" + }, + "invalid pipeline id": func(s *CatalogState) { + s.Outputs[0].PipelineID = ".reports" + }, + "invalid destination id": func(s *CatalogState) { + s.Outputs[0].DestinationID = ".archive" + }, + "missing source id": func(s *CatalogState) { + s.Outputs[0].Source.ID = "" + }, + "invalid source digest": func(s *CatalogState) { + s.Outputs[0].Source.Digest = "SHA256:099b205780d2b050024868399961b05731729a548d5d6329c7b06a6740dd75fe" + }, + "missing source created": func(s *CatalogState) { + s.Outputs[0].Source.Created = time.Time{} + }, + "invalid kind": func(s *CatalogState) { + s.Outputs[0].Kind = "document" + }, + "generated missing source path": func(s *CatalogState) { + s.Outputs[1].SourcePath = "" + }, + "generated invalid source path": func(s *CatalogState) { + s.Outputs[1].SourcePath = "../report.md" + }, + "generated missing transform": func(s *CatalogState) { + s.Outputs[1].Transform = "" + }, + "source output source path": func(s *CatalogState) { + s.Outputs[0].SourcePath = "report.md" + }, + "source output transform": func(s *CatalogState) { + s.Outputs[0].Transform = "markdown_to_html" + }, + "invalid output digest": func(s *CatalogState) { + s.Outputs[0].SHA256 = "SHA256:3640fd37140ee4d2e0e93e78834f232ea67a50e7bc6279203690cc7de1975fa6" + }, + "negative size": func(s *CatalogState) { + s.Outputs[0].Size = -1 + }, + "invalid url": func(s *CatalogState) { + s.Outputs[1].URL = "file:///tmp/report.html" + }, + "missing created at": func(s *CatalogState) { + s.Outputs[0].CreatedAt = time.Time{} + }, + "missing updated at": func(s *CatalogState) { + s.Outputs[0].UpdatedAt = time.Time{} + }, + } + for name, mutate := range tests { + t.Run(name, func(t *testing.T) { + state := validCatalogState(t) + mutate(&state) + if err := ValidateCatalog(state); err == nil { + t.Fatal("ValidateCatalog() error = nil, want error") + } + }) + } +} + +func TestParseCatalogRejectsForbiddenFields(t *testing.T) { + tests := map[string]string{ + "owners": `"owners": [],`, + "sources": `"sources": [],`, + "workflow": `"workflow": "additive",`, + "source": `"source": {"manifest": {}},`, + "manifest": `"manifest": {},`, + "pipeline_id": `"pipeline_id": "reports",`, + "destination_id": `"destination_id": "archive",`, + "published_at": `"published_at": "2026-06-19T12:00:00Z",`, + } + for name, field := range tests { + t.Run(name, func(t *testing.T) { + body := strings.Replace(validCatalogStateJSON(t), `"created_at":`, field+"\n "+`"created_at":`, 1) + _, err := ParseCatalog([]byte(body)) + assertStateErrorContains(t, err, "unknown field") + }) + } +} + +func TestParseCatalogRejectsForbiddenOutputFieldsForSourceOutput(t *testing.T) { + tests := map[string]string{ + "source_path": `"source_path": "report.md",`, + "transform": `"transform": "markdown_to_html",`, + "empty url": `"url": "",`, + } + for name, field := range tests { + t.Run(name, func(t *testing.T) { + body := strings.Replace(validCatalogStateJSON(t), `"kind": "source",`, `"kind": "source",`+"\n "+field, 1) + _, err := ParseCatalog([]byte(body)) + if err == nil { + t.Fatal("ParseCatalog() error = nil, want error") + } + }) + } +} + +func validCatalogStateJSON(t *testing.T) string { + t.Helper() + data, err := json.MarshalIndent(validCatalogState(t), "", " ") + if err != nil { + t.Fatalf("marshal catalog state: %v", err) + } + return string(data) +} + +func catalogStateObject(t *testing.T) map[string]any { + t.Helper() + var document map[string]any + if err := json.Unmarshal([]byte(validCatalogStateJSON(t)), &document); err != nil { + t.Fatalf("unmarshal catalog state: %v", err) + } + return document +} + +func firstCatalogOutput(document map[string]any) map[string]any { + outputs := document["outputs"].([]any) + return outputs[0].(map[string]any) +} + +func firstCatalogSource(document map[string]any) map[string]any { + return firstCatalogOutput(document)["source"].(map[string]any) +} + +func mustMarshalCatalogObject(t *testing.T, document map[string]any) []byte { + t.Helper() + data, err := json.Marshal(document) + if err != nil { + t.Fatalf("marshal catalog object: %v", err) + } + return data +} + +func validCatalogState(t *testing.T) CatalogState { + t.Helper() + manifest := validManifest(t) + createdAt := time.Date(2026, 6, 19, 12, 0, 0, 0, time.UTC) + updatedAt := time.Date(2026, 6, 19, 12, 5, 0, 0, time.UTC) + source := CatalogSourceIdentity{ + ID: manifest.ID, + Digest: manifest.Digest, + Created: manifest.Created, + } + return CatalogState{ + SchemaVersion: CatalogSchemaVersion, + DistributorVersion: "dev", + CreatedAt: createdAt, + UpdatedAt: updatedAt, + State: StatePolicy{Mode: StateModeCatalog}, + Outputs: []CatalogOutputFile{{ + Path: "report.md", + PipelineID: "reports", + DestinationID: "archive", + Source: source, + Kind: OutputKindSource, + SHA256: manifest.Files[0].SHA256, + Size: manifest.Files[0].Size, + CreatedAt: createdAt, + UpdatedAt: updatedAt, + }, { + Path: "report.html", + PipelineID: "reports", + DestinationID: "html", + Source: source, + Kind: OutputKindGenerated, + SourcePath: "report.md", + Transform: "markdown_to_html", + URL: "https://reports.example.com/report.html", + SHA256: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + Size: 128, + CreatedAt: createdAt, + UpdatedAt: updatedAt, + }}, + } +} diff --git a/internal/state/compare.go b/internal/state/compare.go index 475448e..a03df35 100644 --- a/internal/state/compare.go +++ b/internal/state/compare.go @@ -21,10 +21,12 @@ const ( ) type DestinationStatus struct { - State *DistributorState - SharedRoot *SharedRootState - StateErr error - HasContents bool + State *DistributorState + SharedRoot *SharedRootState + Catalog *CatalogState + SupersededLegacy *SupersededLegacyState + StateErr error + HasContents bool } type Comparison struct { diff --git a/internal/state/distributor.go b/internal/state/distributor.go index 125da7f..8a14d53 100644 --- a/internal/state/distributor.go +++ b/internal/state/distributor.go @@ -14,9 +14,11 @@ import ( const ( SchemaVersion = 2 SharedRootSchemaVersion = 3 + CatalogSchemaVersion = 4 legacySchemaVersion = 1 StateModeSingleOwner = config.StateModeSingleOwner StateModeSharedRoot = config.StateModeSharedRoot + StateModeCatalog = "catalog" ) type DistributorState struct { diff --git a/internal/state/shared_root.go b/internal/state/shared_root.go index c70ba2f..7f588e2 100644 --- a/internal/state/shared_root.go +++ b/internal/state/shared_root.go @@ -14,8 +14,14 @@ import ( ) type StateDocument struct { - SingleOwner *DistributorState - SharedRoot *SharedRootState + SingleOwner *DistributorState + SharedRoot *SharedRootState + Catalog *CatalogState + SupersededLegacy *SupersededLegacyState +} + +type SupersededLegacyState struct { + SchemaVersion int } type SharedRootState struct { @@ -103,18 +109,18 @@ func ParseDocument(data []byte) (StateDocument, error) { if err != nil { return StateDocument{}, err } - if schemaVersion == SharedRootSchemaVersion { - sharedRoot, err := ParseSharedRoot(data) + switch schemaVersion { + case legacySchemaVersion, SchemaVersion, SharedRootSchemaVersion: + return StateDocument{SupersededLegacy: &SupersededLegacyState{SchemaVersion: schemaVersion}}, nil + case CatalogSchemaVersion: + catalog, err := ParseCatalog(data) if err != nil { return StateDocument{}, err } - return StateDocument{SharedRoot: &sharedRoot}, nil + return StateDocument{Catalog: &catalog}, nil + default: + return StateDocument{}, fmt.Errorf("state schema_version %d is unsupported", schemaVersion) } - singleOwner, err := Parse(data) - if err != nil { - return StateDocument{}, err - } - return StateDocument{SingleOwner: &singleOwner}, nil } func parseSchemaVersion(data []byte) (int, error) { @@ -125,6 +131,10 @@ func parseSchemaVersion(data []byte) (int, error) { if err := decoder.Decode(&raw); err != nil { return 0, fmt.Errorf("parse distributor state: %w", err) } + var extra any + if err := decoder.Decode(&extra); err != io.EOF { + return 0, fmt.Errorf("parse distributor state: trailing data") + } if raw.SchemaVersion == nil { return 0, fmt.Errorf("state schema_version is required") } diff --git a/internal/state/shared_root_test.go b/internal/state/shared_root_test.go index e447973..4573ec2 100644 --- a/internal/state/shared_root_test.go +++ b/internal/state/shared_root_test.go @@ -10,24 +10,6 @@ import ( "gitea.maximumdirect.net/eric/distributor/internal/config" ) -func TestParseDocumentHandlesSingleOwnerAndSharedRoot(t *testing.T) { - singleOwner, err := ParseDocument([]byte(validStateJSON(t))) - if err != nil { - t.Fatalf("ParseDocument(single owner) error = %v", err) - } - if singleOwner.SingleOwner == nil || singleOwner.SharedRoot != nil { - t.Fatalf("single owner document = %#v", singleOwner) - } - - sharedRoot, err := ParseDocument([]byte(validSharedRootStateJSON(t))) - if err != nil { - t.Fatalf("ParseDocument(shared root) error = %v", err) - } - if sharedRoot.SharedRoot == nil || sharedRoot.SingleOwner != nil { - t.Fatalf("shared root document = %#v", sharedRoot) - } -} - func TestParseSharedRootState(t *testing.T) { state, err := ParseSharedRoot([]byte(validSharedRootStateJSON(t))) if err != nil {