Add destination state comparison

2026-05-31 02:13:58 +00:00
parent 518944e601
commit aeee91940d
6 changed files with 762 additions and 0 deletions
--- a/docs/internal/state.md
+++ b/docs/internal/state.md
@@ -0,0 +1,40 @@
+# Destination State
+
+## Purpose
+
+`internal/state` parses, validates, writes, and compares `.distributor.json` destination state.
+
+## Inputs and outputs
+
+Input is JSON destination state plus the current source manifest, pipeline id, destination id, and whether the destination path has unmanaged content. Output is a deterministic comparison outcome and reason.
+
+## State behavior
+
+`.distributor.json` requires:
+
+- `schema_version: 1`
+- `pipeline_id`
+- `destination_id`
+- `published_at`
+- `source.manifest`
+- `outputs`
+
+`distributor_version` is optional diagnostic metadata. `published_at` parses as RFC3339 and distributor-written state serializes it as RFC3339 UTC.
+
+The embedded `source.manifest` is validated with the same source manifest rules used by `internal/bundle`.
+
+## Outputs
+
+Each output records `path`, `kind`, `source_path`, `sha256`, and `size`. Supported output kinds are `source` and `generated`. Generated outputs require `transform`.
+
+## Comparison
+
+Comparison outcomes cover absent destination state, unmanaged destination content, invalid state, pipeline or destination mismatch, same source manifest, older destination source, newer destination source, same-created digest conflict, and different source id conflict.
+
+## Boundaries
+
+This package does not publish files, delete files, inspect storage backends, or choose transfer policy actions. Publish planning consumes these comparison outcomes later.
+
+## Tests
+
+Before changing destination state behavior, inspect tests under `internal/state`.
--- a/internal/state/compare.go
+++ b/internal/state/compare.go
@@ -0,0 +1,89 @@
+package state
+
+import (
+	"fmt"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+)
+
+type Outcome string
+
+const (
+	OutcomeDestinationAbsent       Outcome = "destination_absent"
+	OutcomeDestinationUnmanaged    Outcome = "destination_unmanaged"
+	OutcomeInvalidState            Outcome = "invalid_destination_state"
+	OutcomeIdentityMismatch        Outcome = "destination_identity_mismatch"
+	OutcomeSameSource              Outcome = "same_source_manifest"
+	OutcomeDestinationOlder        Outcome = "destination_older"
+	OutcomeDestinationNewer        Outcome = "destination_newer"
+	OutcomeSameCreatedConflict     Outcome = "same_created_digest_conflict"
+	OutcomeDifferentSourceConflict Outcome = "different_source_conflict"
+)
+
+type DestinationStatus struct {
+	State       *DistributorState
+	StateErr    error
+	HasContents bool
+}
+
+type Comparison struct {
+	Outcome Outcome
+	Reason  string
+}
+
+func Compare(source bundle.Manifest, pipelineID, destinationID string, status DestinationStatus) Comparison {
+	if status.StateErr != nil {
+		return Comparison{Outcome: OutcomeInvalidState, Reason: status.StateErr.Error()}
+	}
+	if status.State == nil {
+		if status.HasContents {
+			return Comparison{Outcome: OutcomeDestinationUnmanaged, Reason: "destination has content but no distributor state"}
+		}
+		return Comparison{Outcome: OutcomeDestinationAbsent, Reason: "destination state is absent"}
+	}
+
+	destinationState := *status.State
+	if err := Validate(destinationState); err != nil {
+		return Comparison{Outcome: OutcomeInvalidState, Reason: err.Error()}
+	}
+	if destinationState.PipelineID != pipelineID {
+		return Comparison{Outcome: OutcomeIdentityMismatch, Reason: fmt.Sprintf("pipeline id %q does not match %q", destinationState.PipelineID, pipelineID)}
+	}
+	if destinationState.DestinationID != destinationID {
+		return Comparison{Outcome: OutcomeIdentityMismatch, Reason: fmt.Sprintf("destination id %q does not match %q", destinationState.DestinationID, destinationID)}
+	}
+
+	destinationManifest := destinationState.Source.Manifest
+	if manifestsEqual(source, destinationManifest) {
+		return Comparison{Outcome: OutcomeSameSource, Reason: "destination source manifest matches source"}
+	}
+	if destinationManifest.ID != source.ID {
+		return Comparison{Outcome: OutcomeDifferentSourceConflict, Reason: "destination source id differs from source"}
+	}
+	if destinationManifest.Created.Before(source.Created) {
+		return Comparison{Outcome: OutcomeDestinationOlder, Reason: "destination source is older than source"}
+	}
+	if destinationManifest.Created.After(source.Created) {
+		return Comparison{Outcome: OutcomeDestinationNewer, Reason: "destination source is newer than source"}
+	}
+	if destinationManifest.Digest != source.Digest {
+		return Comparison{Outcome: OutcomeSameCreatedConflict, Reason: "destination source has same id and created time but different digest"}
+	}
+	return Comparison{Outcome: OutcomeInvalidState, Reason: "destination source differs from source without a supported comparison outcome"}
+}
+
+func manifestsEqual(a, b bundle.Manifest) bool {
+	if a.SchemaVersion != b.SchemaVersion ||
+		a.ID != b.ID ||
+		a.Digest != b.Digest ||
+		!a.Created.Equal(b.Created) ||
+		len(a.Files) != len(b.Files) {
+		return false
+	}
+	for index := range a.Files {
+		if a.Files[index] != b.Files[index] {
+			return false
+		}
+	}
+	return true
+}
--- a/internal/state/compare_test.go
+++ b/internal/state/compare_test.go
@@ -0,0 +1,121 @@
+package state
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+)
+
+func TestCompareOutcomes(t *testing.T) {
+	source := validManifest(t)
+	tests := []struct {
+		name   string
+		status DestinationStatus
+		want   Outcome
+	}{
+		{
+			name:   "destination absent",
+			status: DestinationStatus{},
+			want:   OutcomeDestinationAbsent,
+		},
+		{
+			name:   "destination unmanaged",
+			status: DestinationStatus{HasContents: true},
+			want:   OutcomeDestinationUnmanaged,
+		},
+		{
+			name:   "invalid destination state",
+			status: DestinationStatus{StateErr: errors.New("invalid json")},
+			want:   OutcomeInvalidState,
+		},
+		{
+			name:   "pipeline mismatch",
+			status: DestinationStatus{State: withState(t, source, func(s *DistributorState) { s.PipelineID = "other" })},
+			want:   OutcomeIdentityMismatch,
+		},
+		{
+			name:   "destination mismatch",
+			status: DestinationStatus{State: withState(t, source, func(s *DistributorState) { s.DestinationID = "other" })},
+			want:   OutcomeIdentityMismatch,
+		},
+		{
+			name:   "same source manifest",
+			status: DestinationStatus{State: withState(t, source, nil)},
+			want:   OutcomeSameSource,
+		},
+		{
+			name: "destination older",
+			status: DestinationStatus{State: withState(t, source, func(s *DistributorState) {
+				s.Source.Manifest.Created = source.Created.Add(-time.Hour)
+			})},
+			want: OutcomeDestinationOlder,
+		},
+		{
+			name: "destination newer",
+			status: DestinationStatus{State: withState(t, source, func(s *DistributorState) {
+				s.Source.Manifest.Created = source.Created.Add(time.Hour)
+			})},
+			want: OutcomeDestinationNewer,
+		},
+		{
+			name: "same created digest conflict",
+			status: DestinationStatus{State: withState(t, source, func(s *DistributorState) {
+				s.Source.Manifest.Files[0].SHA256 = "sha256:3333333333333333333333333333333333333333333333333333333333333333"
+				s.Source.Manifest.Digest = bundle.BundleDigest(s.Source.Manifest.Files)
+			})},
+			want: OutcomeSameCreatedConflict,
+		},
+		{
+			name: "different source id",
+			status: DestinationStatus{State: withState(t, source, func(s *DistributorState) {
+				s.Source.Manifest.ID = "other.source"
+			})},
+			want: OutcomeDifferentSourceConflict,
+		},
+		{
+			name: "invalid state object",
+			status: DestinationStatus{State: withState(t, source, func(s *DistributorState) {
+				s.Outputs[0].Kind = "other"
+			})},
+			want: OutcomeInvalidState,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := Compare(source, "reports", "archive", tt.status)
+			if got.Outcome != tt.want {
+				t.Fatalf("Compare() outcome = %s reason=%q, want %s", got.Outcome, got.Reason, tt.want)
+			}
+			if got.Reason == "" {
+				t.Fatal("Compare() reason is empty")
+			}
+		})
+	}
+}
+
+func withState(t *testing.T, source bundle.Manifest, mutate func(*DistributorState)) *DistributorState {
+	t.Helper()
+	stateManifest := source
+	stateManifest.Files = append([]bundle.ManifestFile(nil), source.Files...)
+	state := DistributorState{
+		SchemaVersion: SchemaVersion,
+		PipelineID:    "reports",
+		DestinationID: "archive",
+		PublishedAt:   time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC),
+		Source:        SourceState{Manifest: stateManifest},
+		Outputs: []OutputFile{{
+			Path:       "report.md",
+			Kind:       OutputKindSource,
+			SourcePath: "report.md",
+			SHA256:     source.Files[0].SHA256,
+			Size:       source.Files[0].Size,
+		}},
+	}
+	if mutate != nil {
+		mutate(&state)
+	}
+	return &state
+}
--- a/internal/state/distributor.go
+++ b/internal/state/distributor.go
@@ -0,0 +1,214 @@
+package state
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+)
+
+const SchemaVersion = 1
+
+type DistributorState struct {
+	SchemaVersion      int
+	DistributorVersion string
+	PipelineID         string
+	DestinationID      string
+	PublishedAt        time.Time
+	Source             SourceState
+	Outputs            []OutputFile
+}
+
+type SourceState struct {
+	Manifest bundle.Manifest
+}
+
+type OutputFile struct {
+	Path       string
+	Kind       string
+	SourcePath string
+	Transform  string
+	SHA256     string
+	Size       int64
+}
+
+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"`
+	Outputs            []rawOutputFile `json:"outputs"`
+}
+
+type rawSourceState struct {
+	Manifest json.RawMessage `json:"manifest"`
+}
+
+type rawOutputFile struct {
+	Path       *string `json:"path"`
+	Kind       *string `json:"kind"`
+	SourcePath *string `json:"source_path"`
+	Transform  string  `json:"transform"`
+	SHA256     *string `json:"sha256"`
+	Size       *int64  `json:"size"`
+}
+
+func Parse(data []byte) (DistributorState, error) {
+	decoder := json.NewDecoder(bytes.NewReader(data))
+	var raw rawDistributorState
+	if err := decoder.Decode(&raw); err != nil {
+		return DistributorState{}, fmt.Errorf("parse distributor state: %w", err)
+	}
+	var extra any
+	if err := decoder.Decode(&extra); err != io.EOF {
+		return DistributorState{}, fmt.Errorf("parse distributor state: trailing data")
+	}
+	state, err := parseRaw(raw)
+	if err != nil {
+		return DistributorState{}, err
+	}
+	if err := Validate(state); err != nil {
+		return DistributorState{}, err
+	}
+	return state, nil
+}
+
+func parseRaw(raw rawDistributorState) (DistributorState, error) {
+	var state DistributorState
+	if raw.SchemaVersion == nil {
+		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)
+	}
+	state.DistributorVersion = raw.DistributorVersion
+	if raw.PipelineID == nil || *raw.PipelineID == "" {
+		return DistributorState{}, fmt.Errorf("state pipeline_id is required")
+	}
+	state.PipelineID = *raw.PipelineID
+	if raw.DestinationID == nil || *raw.DestinationID == "" {
+		return DistributorState{}, fmt.Errorf("state destination_id is required")
+	}
+	state.DestinationID = *raw.DestinationID
+	if raw.PublishedAt == nil || *raw.PublishedAt == "" {
+		return DistributorState{}, fmt.Errorf("state published_at is required")
+	}
+	publishedAt, err := time.Parse(time.RFC3339, *raw.PublishedAt)
+	if err != nil {
+		return DistributorState{}, fmt.Errorf("state published_at must be RFC3339: %w", err)
+	}
+	state.PublishedAt = publishedAt.UTC()
+	if raw.Source == nil || len(raw.Source.Manifest) == 0 {
+		return DistributorState{}, fmt.Errorf("state source.manifest is required")
+	}
+	manifest, err := bundle.ParseManifest(raw.Source.Manifest)
+	if err != nil {
+		return DistributorState{}, fmt.Errorf("state source.manifest: %w", err)
+	}
+	state.Source.Manifest = manifest
+	if raw.Outputs == nil {
+		return DistributorState{}, fmt.Errorf("state outputs is required")
+	}
+	outputs, err := parseOutputs(raw.Outputs)
+	if err != nil {
+		return DistributorState{}, err
+	}
+	state.Outputs = outputs
+	return state, nil
+}
+
+func parseOutputs(rawOutputs []rawOutputFile) ([]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)
+		if err != nil {
+			return nil, err
+		}
+		if _, exists := seen[output.Path]; exists {
+			return nil, fmt.Errorf("state outputs[%d].path duplicates %q", index, output.Path)
+		}
+		seen[output.Path] = struct{}{}
+		outputs = append(outputs, output)
+	}
+	return outputs, nil
+}
+
+func parseOutput(index int, raw rawOutputFile) (OutputFile, error) {
+	if raw.Path == nil || *raw.Path == "" {
+		return OutputFile{}, fmt.Errorf("state outputs[%d].path is required", index)
+	}
+	if raw.Kind == nil || *raw.Kind == "" {
+		return OutputFile{}, fmt.Errorf("state outputs[%d].kind is required", index)
+	}
+	if raw.SourcePath == nil || *raw.SourcePath == "" {
+		return OutputFile{}, fmt.Errorf("state outputs[%d].source_path is required", index)
+	}
+	if raw.SHA256 == nil || *raw.SHA256 == "" {
+		return OutputFile{}, fmt.Errorf("state outputs[%d].sha256 is required", index)
+	}
+	if raw.Size == nil {
+		return OutputFile{}, fmt.Errorf("state outputs[%d].size is required", index)
+	}
+	return OutputFile{
+		Path:       *raw.Path,
+		Kind:       *raw.Kind,
+		SourcePath: *raw.SourcePath,
+		Transform:  raw.Transform,
+		SHA256:     *raw.SHA256,
+		Size:       *raw.Size,
+	}, nil
+}
+
+func (s DistributorState) PublishedAtString() string {
+	return s.PublishedAt.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"`
+		Outputs            []OutputFile `json:"outputs"`
+	}
+	return json.Marshal(stateJSON{
+		SchemaVersion:      s.SchemaVersion,
+		DistributorVersion: s.DistributorVersion,
+		PipelineID:         s.PipelineID,
+		DestinationID:      s.DestinationID,
+		PublishedAt:        s.PublishedAtString(),
+		Source:             sourceJSON{Manifest: s.Source.Manifest},
+		Outputs:            s.Outputs,
+	})
+}
+
+func (o OutputFile) MarshalJSON() ([]byte, error) {
+	type outputJSON struct {
+		Path       string `json:"path"`
+		Kind       string `json:"kind"`
+		SourcePath string `json:"source_path"`
+		Transform  string `json:"transform,omitempty"`
+		SHA256     string `json:"sha256"`
+		Size       int64  `json:"size"`
+	}
+	return json.Marshal(outputJSON{
+		Path:       o.Path,
+		Kind:       o.Kind,
+		SourcePath: o.SourcePath,
+		Transform:  o.Transform,
+		SHA256:     o.SHA256,
+		Size:       o.Size,
+	})
+}
--- a/internal/state/distributor_test.go
+++ b/internal/state/distributor_test.go
@@ -0,0 +1,191 @@
+package state
+
+import (
+	"encoding/json"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+)
+
+func TestParseValidState(t *testing.T) {
+	state, err := Parse([]byte(validStateJSON(t)))
+	if err != nil {
+		t.Fatalf("Parse() error = %v", err)
+	}
+	if state.SchemaVersion != SchemaVersion {
+		t.Fatalf("schema version = %d, want %d", state.SchemaVersion, SchemaVersion)
+	}
+	if state.PipelineID != "reports" || state.DestinationID != "archive" {
+		t.Fatalf("identity = %q/%q", state.PipelineID, state.DestinationID)
+	}
+	if got, want := state.PublishedAtString(), "2026-05-30T11:12:00Z"; got != want {
+		t.Fatalf("PublishedAtString() = %q, want %q", got, want)
+	}
+	if got, want := len(state.Outputs), 1; got != want {
+		t.Fatalf("output count = %d, want %d", got, want)
+	}
+}
+
+func TestParseNormalizesPublishedAtOffset(t *testing.T) {
+	body := strings.Replace(validStateJSON(t), `"published_at": "2026-05-30T11:12:00Z"`, `"published_at": "2026-05-30T13:12:00+02:00"`, 1)
+	state, err := Parse([]byte(body))
+	if err != nil {
+		t.Fatalf("Parse() error = %v", err)
+	}
+	if got, want := state.PublishedAtString(), "2026-05-30T11:12:00Z"; got != want {
+		t.Fatalf("PublishedAtString() = %q, want %q", got, want)
+	}
+}
+
+func TestParseRejectsMissingFields(t *testing.T) {
+	tests := map[string]string{
+		"schema_version": `"schema_version"`,
+		"pipeline_id":    `"pipeline_id"`,
+		"destination_id": `"destination_id"`,
+		"published_at":   `"published_at"`,
+		"source":         `"source"`,
+		"outputs":        `"outputs"`,
+	}
+	for name, field := range tests {
+		t.Run(name, func(t *testing.T) {
+			body := strings.Replace(validStateJSON(t), field, `"missing_`+name+`"`, 1)
+			_, err := Parse([]byte(body))
+			assertStateErrorContains(t, err, "required")
+		})
+	}
+}
+
+func TestParseRejectsInvalidSchemaVersion(t *testing.T) {
+	body := strings.Replace(validStateJSON(t), `"schema_version": 1`, `"schema_version": 2`, 1)
+	_, err := Parse([]byte(body))
+	assertStateErrorContains(t, err, "schema_version must be 1")
+}
+
+func TestParseRejectsInvalidEmbeddedManifest(t *testing.T) {
+	body := validStateWithManifestJSON(t, strings.Replace(manifestJSON(t), `"schema_version": 1`, `"schema_version": 2`, 1))
+	_, err := Parse([]byte(body))
+	assertStateErrorContains(t, err, "source.manifest")
+}
+
+func TestParseRejectsInvalidOutputMetadata(t *testing.T) {
+	source := validManifest(t)
+	tests := map[string]func(*DistributorState){
+		"unsafe path": func(s *DistributorState) {
+			s.Outputs[0].Path = "../report.md"
+		},
+		"invalid kind": func(s *DistributorState) {
+			s.Outputs[0].Kind = "other"
+		},
+		"invalid source": func(s *DistributorState) {
+			s.Outputs[0].SourcePath = "../report.md"
+		},
+		"generated missing": func(s *DistributorState) {
+			s.Outputs[0].Kind = OutputKindGenerated
+		},
+		"invalid digest": func(s *DistributorState) {
+			s.Outputs[0].SHA256 = "SHA256:3640fd37140ee4d2e0e93e78834f232ea67a50e7bc6279203690cc7de1975fa6"
+		},
+		"negative size": func(s *DistributorState) {
+			s.Outputs[0].Size = -1
+		},
+	}
+	for name, mutate := range tests {
+		t.Run(name, func(t *testing.T) {
+			state := *withState(t, source, mutate)
+			err := Validate(state)
+			if err == nil {
+				t.Fatal("Validate() error = nil, want error")
+			}
+		})
+	}
+}
+
+func TestParseRejectsMalformedPublishedTimestamp(t *testing.T) {
+	body := strings.Replace(validStateJSON(t), `"published_at": "2026-05-30T11:12:00Z"`, `"published_at": "May 30"`, 1)
+	_, err := Parse([]byte(body))
+	assertStateErrorContains(t, err, "published_at must be RFC3339")
+}
+
+func TestMarshalNormalizesPublishedAtUTC(t *testing.T) {
+	source := validManifest(t)
+	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},
+		Outputs: []OutputFile{{
+			Path:       "report.md",
+			Kind:       OutputKindSource,
+			SourcePath: "report.md",
+			SHA256:     source.Files[0].SHA256,
+			Size:       source.Files[0].Size,
+		}},
+	}
+	data, err := json.Marshal(state)
+	if err != nil {
+		t.Fatalf("Marshal() error = %v", err)
+	}
+	if !strings.Contains(string(data), `"published_at":"2026-05-30T11:12:00Z"`) {
+		t.Fatalf("json = %s, want UTC RFC3339 published_at", data)
+	}
+}
+
+func validStateJSON(t *testing.T) string {
+	t.Helper()
+	return validStateWithManifestJSON(t, manifestJSON(t))
+}
+
+func validStateWithManifestJSON(t *testing.T, manifest string) string {
+	t.Helper()
+	return `{
+  "schema_version": 1,
+  "distributor_version": "dev",
+  "pipeline_id": "reports",
+  "destination_id": "archive",
+  "published_at": "2026-05-30T11:12:00Z",
+  "source": {
+    "manifest": ` + manifest + `
+  },
+  "outputs": [
+    {
+      "path": "report.md",
+      "kind": "source",
+      "source_path": "report.md",
+      "sha256": "sha256:3640fd37140ee4d2e0e93e78834f232ea67a50e7bc6279203690cc7de1975fa6",
+      "size": 16
+    }
+  ]
+}`
+}
+
+func manifestJSON(t *testing.T) string {
+	t.Helper()
+	data, err := os.ReadFile("../bundle/testdata/valid_bundle/manifest.json")
+	if err != nil {
+		t.Fatalf("read manifest fixture: %v", err)
+	}
+	return string(data)
+}
+
+func validManifest(t *testing.T) bundle.Manifest {
+	t.Helper()
+	manifest, err := bundle.ParseManifest([]byte(manifestJSON(t)))
+	if err != nil {
+		t.Fatalf("ParseManifest() error = %v", err)
+	}
+	return manifest
+}
+
+func assertStateErrorContains(t *testing.T, err error, want string) {
+	t.Helper()
+	if err == nil {
+		t.Fatalf("error = nil, want substring %q", want)
+	}
+	if !strings.Contains(err.Error(), want) {
+		t.Fatalf("error = %q, want substring %q", err.Error(), want)
+	}
+}
--- a/internal/state/validate.go
+++ b/internal/state/validate.go
@@ -0,0 +1,107 @@
+package state
+
+import (
+	"fmt"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+)
+
+const (
+	OutputKindSource    = "source"
+	OutputKindGenerated = "generated"
+)
+
+func Validate(s DistributorState) error {
+	if s.SchemaVersion != SchemaVersion {
+		return fmt.Errorf("state schema_version must be %d", SchemaVersion)
+	}
+	if s.PipelineID == "" {
+		return fmt.Errorf("state pipeline_id is required")
+	}
+	if s.DestinationID == "" {
+		return fmt.Errorf("state destination_id is required")
+	}
+	if s.PublishedAt.IsZero() {
+		return fmt.Errorf("state published_at is required")
+	}
+	if err := validateEmbeddedManifest(s.Source.Manifest); err != nil {
+		return fmt.Errorf("state source.manifest: %w", err)
+	}
+	if s.Outputs == nil {
+		return fmt.Errorf("state outputs is required")
+	}
+	seen := make(map[string]struct{}, len(s.Outputs))
+	for index, output := range s.Outputs {
+		if err := validateOutput(index, output); err != nil {
+			return err
+		}
+		if _, exists := seen[output.Path]; exists {
+			return fmt.Errorf("state outputs[%d].path duplicates %q", index, output.Path)
+		}
+		seen[output.Path] = struct{}{}
+	}
+	return nil
+}
+
+func validateEmbeddedManifest(manifest bundle.Manifest) error {
+	if manifest.SchemaVersion != 1 {
+		return fmt.Errorf("schema_version must be 1")
+	}
+	if manifest.ID == "" {
+		return fmt.Errorf("id is required")
+	}
+	if err := bundle.ValidateDigest(manifest.Digest); err != nil {
+		return fmt.Errorf("digest: %w", err)
+	}
+	if manifest.Created.IsZero() {
+		return fmt.Errorf("created is required")
+	}
+	if len(manifest.Files) == 0 {
+		return fmt.Errorf("files is required")
+	}
+	seen := make(map[string]struct{}, len(manifest.Files))
+	for index, file := range manifest.Files {
+		if err := bundle.ValidateSourcePath(file.Path); err != nil {
+			return fmt.Errorf("files[%d].path: %w", index, err)
+		}
+		if err := bundle.ValidateDigest(file.SHA256); err != nil {
+			return fmt.Errorf("files[%d].sha256: %w", index, err)
+		}
+		if file.Size < 0 {
+			return fmt.Errorf("files[%d].size must be non-negative", index)
+		}
+		if _, exists := seen[file.Path]; exists {
+			return fmt.Errorf("files[%d].path duplicates %q", index, file.Path)
+		}
+		seen[file.Path] = struct{}{}
+	}
+	if actual := bundle.BundleDigest(manifest.Files); actual != manifest.Digest {
+		return fmt.Errorf("digest mismatch: got %s want %s", actual, manifest.Digest)
+	}
+	return nil
+}
+
+func validateOutput(index int, output OutputFile) error {
+	if err := storage.ValidatePath(output.Path); err != nil {
+		return fmt.Errorf("state outputs[%d].path: %w", index, err)
+	}
+	switch output.Kind {
+	case OutputKindSource, OutputKindGenerated:
+	default:
+		return fmt.Errorf("state outputs[%d].kind must be source or generated", index)
+	}
+	if err := storage.ValidatePath(output.SourcePath); err != nil {
+		return fmt.Errorf("state outputs[%d].source_path: %w", index, err)
+	}
+	if output.Kind == OutputKindGenerated && output.Transform == "" {
+		return fmt.Errorf("state outputs[%d].transform is required for generated output", index)
+	}
+	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)
+	}
+	return nil
+}