Add catalog destination state schema

2026-06-19 15:28:03 +00:00
parent a95662226f
commit 20129bfff5
9 changed files with 834 additions and 36 deletions
--- 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 {
--- a/internal/app/reconcile_state.go
+++ b/internal/app/reconcile_state.go
@@ -161,8 +161,11 @@ func buildReconcileStateReport(ctx context.Context, backend storage.Backend, pip
 	if document.SingleOwner != nil {
 		return reconcileSingleOwnerState(ctx, backend, statePath, *document.SingleOwner, 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) {
 	if destinationState.PipelineID != scope.PipelineID || destinationState.DestinationID != scope.DestinationID {
--- 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
--- a/internal/state/catalog.go
+++ 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
+}
--- a/internal/state/catalog_test.go
+++ 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,
+		}},
+	}
+}
--- a/internal/state/compare.go
+++ b/internal/state/compare.go
@@ -23,6 +23,8 @@ const (
 type DestinationStatus struct {
 	State            *DistributorState
 	SharedRoot       *SharedRootState
+	Catalog          *CatalogState
+	SupersededLegacy *SupersededLegacyState
 	StateErr         error
 	HasContents      bool
 }
--- 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 {
--- a/internal/state/shared_root.go
+++ b/internal/state/shared_root.go
@@ -16,6 +16,12 @@ import (
 type StateDocument struct {
 	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")
 	}
--- 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 {