distributor/internal/state/shared_root.go

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 StateDocument struct {
	SingleOwner *DistributorState
	SharedRoot  *SharedRootState
}

type SharedRootState struct {
	SchemaVersion      int
	DistributorVersion string
	CreatedAt          time.Time
	UpdatedAt          time.Time
	State              StatePolicy
	Owners             []OwnerRecord
	Outputs            []SharedRootOutputFile
}

type OwnerScope struct {
	PipelineID    string
	DestinationID string
}

type OwnerRecord struct {
	Scope          OwnerScope
	Reconciliation ReconciliationPolicy
	Source         SourceState
	Links          *LinkState
}

type SharedRootOutputFile struct {
	Path          string
	Kind          string
	SourcePath    string
	Transform     string
	URL           string
	SHA256        string
	Size          int64
	Owner         OwnerScope
	SourceID      string
	SourceDigest  string
	SourceCreated time.Time
	CreatedAt     time.Time
	UpdatedAt     time.Time
}

type PathOwnershipConflict struct {
	Path  string
	Owner OwnerScope
}

type rawSharedRootState 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"`
	Owners             []rawOwnerRecord      `json:"owners"`
	Outputs            []rawSharedRootOutput `json:"outputs"`
}

type rawOwnerRecord struct {
	PipelineID     *string                  `json:"pipeline_id"`
	DestinationID  *string                  `json:"destination_id"`
	Reconciliation *rawReconciliationPolicy `json:"reconciliation"`
	Source         *rawSourceState          `json:"source"`
	Links          *rawLinkState            `json:"links"`
}

type rawSharedRootOutput struct {
	Path          *string `json:"path"`
	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"`
	PipelineID    *string `json:"pipeline_id"`
	DestinationID *string `json:"destination_id"`
	SourceID      *string `json:"source_id"`
	SourceDigest  *string `json:"source_digest"`
	SourceCreated *string `json:"source_created"`
	CreatedAt     *string `json:"created_at"`
	UpdatedAt     *string `json:"updated_at"`
}

func ParseDocument(data []byte) (StateDocument, error) {
	schemaVersion, err := parseSchemaVersion(data)
	if err != nil {
		return StateDocument{}, err
	}
	if schemaVersion == SharedRootSchemaVersion {
		sharedRoot, err := ParseSharedRoot(data)
		if err != nil {
			return StateDocument{}, err
		}
		return StateDocument{SharedRoot: &sharedRoot}, nil
	}
	singleOwner, err := Parse(data)
	if err != nil {
		return StateDocument{}, err
	}
	return StateDocument{SingleOwner: &singleOwner}, nil
}

func parseSchemaVersion(data []byte) (int, error) {
	decoder := json.NewDecoder(bytes.NewReader(data))
	var raw struct {
		SchemaVersion *int `json:"schema_version"`
	}
	if err := decoder.Decode(&raw); err != nil {
		return 0, fmt.Errorf("parse distributor state: %w", err)
	}
	if raw.SchemaVersion == nil {
		return 0, fmt.Errorf("state schema_version is required")
	}
	return *raw.SchemaVersion, nil
}

func ParseSharedRoot(data []byte) (SharedRootState, error) {
	decoder := json.NewDecoder(bytes.NewReader(data))
	var raw rawSharedRootState
	if err := decoder.Decode(&raw); err != nil {
		return SharedRootState{}, fmt.Errorf("parse distributor state: %w", err)
	}
	var extra any
	if err := decoder.Decode(&extra); err != io.EOF {
		return SharedRootState{}, fmt.Errorf("parse distributor state: trailing data")
	}
	state, err := parseSharedRootRaw(raw)
	if err != nil {
		return SharedRootState{}, err
	}
	if err := ValidateSharedRoot(state); err != nil {
		return SharedRootState{}, err
	}
	return state, nil
}

func parseSharedRootRaw(raw rawSharedRootState) (SharedRootState, error) {
	if raw.SchemaVersion == nil {
		return SharedRootState{}, fmt.Errorf("state schema_version is required")
	}
	state := SharedRootState{SchemaVersion: *raw.SchemaVersion}
	if state.SchemaVersion != SharedRootSchemaVersion {
		return SharedRootState{}, fmt.Errorf("state schema_version must be %d", SharedRootSchemaVersion)
	}
	state.DistributorVersion = raw.DistributorVersion
	createdAt, err := parseRequiredTime("state created_at", raw.CreatedAt)
	if err != nil {
		return SharedRootState{}, err
	}
	updatedAt, err := parseRequiredTime("state updated_at", raw.UpdatedAt)
	if err != nil {
		return SharedRootState{}, err
	}
	state.CreatedAt = createdAt
	state.UpdatedAt = updatedAt
	if raw.State == nil || raw.State.Mode == "" {
		return SharedRootState{}, fmt.Errorf("state state.mode is required")
	}
	state.State.Mode = raw.State.Mode
	if raw.Owners == nil {
		return SharedRootState{}, fmt.Errorf("state owners is required")
	}
	owners, err := parseOwnerRecords(raw.Owners)
	if err != nil {
		return SharedRootState{}, err
	}
	state.Owners = owners
	if raw.Outputs == nil {
		return SharedRootState{}, fmt.Errorf("state outputs is required")
	}
	outputs, err := parseSharedRootOutputs(raw.Outputs)
	if err != nil {
		return SharedRootState{}, err
	}
	state.Outputs = outputs
	return state, nil
}

func parseOwnerRecords(rawOwners []rawOwnerRecord) ([]OwnerRecord, error) {
	owners := make([]OwnerRecord, 0, len(rawOwners))
	for index, raw := range rawOwners {
		owner, err := parseOwnerRecord(index, raw)
		if err != nil {
			return nil, err
		}
		owners = append(owners, owner)
	}
	return owners, nil
}

func parseOwnerRecord(index int, raw rawOwnerRecord) (OwnerRecord, error) {
	if raw.PipelineID == nil || *raw.PipelineID == "" {
		return OwnerRecord{}, fmt.Errorf("state owners[%d].pipeline_id is required", index)
	}
	if raw.DestinationID == nil || *raw.DestinationID == "" {
		return OwnerRecord{}, fmt.Errorf("state owners[%d].destination_id is required", index)
	}
	if raw.Reconciliation == nil || raw.Reconciliation.Mode == "" {
		return OwnerRecord{}, fmt.Errorf("state owners[%d].reconciliation.mode is required", index)
	}
	if raw.Source == nil || len(raw.Source.Manifest) == 0 {
		return OwnerRecord{}, fmt.Errorf("state owners[%d].source.manifest is required", index)
	}
	manifest, err := bundle.ParseManifest(raw.Source.Manifest)
	if err != nil {
		return OwnerRecord{}, fmt.Errorf("state owners[%d].source.manifest: %w", index, err)
	}
	owner := OwnerRecord{
		Scope: OwnerScope{
			PipelineID:    *raw.PipelineID,
			DestinationID: *raw.DestinationID,
		},
		Reconciliation: ReconciliationPolicy{Mode: raw.Reconciliation.Mode},
		Source:         SourceState{Manifest: manifest},
	}
	if raw.Links != nil {
		owner.Links = &LinkState{PrimaryURL: raw.Links.PrimaryURL}
	}
	return owner, nil
}

func parseSharedRootOutputs(rawOutputs []rawSharedRootOutput) ([]SharedRootOutputFile, error) {
	outputs := make([]SharedRootOutputFile, 0, len(rawOutputs))
	for index, raw := range rawOutputs {
		output, err := parseSharedRootOutput(index, raw)
		if err != nil {
			return nil, err
		}
		outputs = append(outputs, output)
	}
	return outputs, nil
}

func parseSharedRootOutput(index int, raw rawSharedRootOutput) (SharedRootOutputFile, error) {
	if raw.Path == nil || *raw.Path == "" {
		return SharedRootOutputFile{}, fmt.Errorf("state outputs[%d].path is required", index)
	}
	if raw.Kind == nil || *raw.Kind == "" {
		return SharedRootOutputFile{}, fmt.Errorf("state outputs[%d].kind is required", index)
	}
	if raw.SourcePath == nil || *raw.SourcePath == "" {
		return SharedRootOutputFile{}, fmt.Errorf("state outputs[%d].source_path is required", index)
	}
	if raw.SHA256 == nil || *raw.SHA256 == "" {
		return SharedRootOutputFile{}, fmt.Errorf("state outputs[%d].sha256 is required", index)
	}
	if raw.Size == nil {
		return SharedRootOutputFile{}, fmt.Errorf("state outputs[%d].size is required", index)
	}
	if raw.PipelineID == nil || *raw.PipelineID == "" {
		return SharedRootOutputFile{}, fmt.Errorf("state outputs[%d].pipeline_id is required", index)
	}
	if raw.DestinationID == nil || *raw.DestinationID == "" {
		return SharedRootOutputFile{}, fmt.Errorf("state outputs[%d].destination_id is required", index)
	}
	if raw.SourceID == nil || *raw.SourceID == "" {
		return SharedRootOutputFile{}, fmt.Errorf("state outputs[%d].source_id is required", index)
	}
	if raw.SourceDigest == nil || *raw.SourceDigest == "" {
		return SharedRootOutputFile{}, fmt.Errorf("state outputs[%d].source_digest is required", index)
	}
	sourceCreated, err := parseRequiredTime(fmt.Sprintf("state outputs[%d].source_created", index), raw.SourceCreated)
	if err != nil {
		return SharedRootOutputFile{}, err
	}
	createdAt, err := parseRequiredTime(fmt.Sprintf("state outputs[%d].created_at", index), raw.CreatedAt)
	if err != nil {
		return SharedRootOutputFile{}, err
	}
	updatedAt, err := parseRequiredTime(fmt.Sprintf("state outputs[%d].updated_at", index), raw.UpdatedAt)
	if err != nil {
		return SharedRootOutputFile{}, err
	}
	return SharedRootOutputFile{
		Path:       *raw.Path,
		Kind:       *raw.Kind,
		SourcePath: *raw.SourcePath,
		Transform:  raw.Transform,
		URL:        raw.URL,
		SHA256:     *raw.SHA256,
		Size:       *raw.Size,
		Owner: OwnerScope{
			PipelineID:    *raw.PipelineID,
			DestinationID: *raw.DestinationID,
		},
		SourceID:      *raw.SourceID,
		SourceDigest:  *raw.SourceDigest,
		SourceCreated: sourceCreated,
		CreatedAt:     createdAt,
		UpdatedAt:     updatedAt,
	}, nil
}

func (s SharedRootState) CreatedAtString() string {
	return s.CreatedAt.UTC().Format(time.RFC3339)
}

func (s SharedRootState) UpdatedAtString() string {
	return s.UpdatedAt.UTC().Format(time.RFC3339)
}

func (o SharedRootOutputFile) SourceCreatedString() string {
	return o.SourceCreated.UTC().Format(time.RFC3339)
}

func (o SharedRootOutputFile) CreatedAtString() string {
	return o.CreatedAt.UTC().Format(time.RFC3339)
}

func (o SharedRootOutputFile) UpdatedAtString() string {
	return o.UpdatedAt.UTC().Format(time.RFC3339)
}

func (s SharedRootState) 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"`
		Owners             []OwnerRecord          `json:"owners"`
		Outputs            []SharedRootOutputFile `json:"outputs"`
	}
	return json.Marshal(stateJSON{
		SchemaVersion:      s.SchemaVersion,
		DistributorVersion: s.DistributorVersion,
		CreatedAt:          s.CreatedAtString(),
		UpdatedAt:          s.UpdatedAtString(),
		State:              s.State,
		Owners:             s.Owners,
		Outputs:            s.Outputs,
	})
}

func (o OwnerRecord) MarshalJSON() ([]byte, error) {
	type sourceJSON struct {
		Manifest bundle.Manifest `json:"manifest"`
	}
	type ownerJSON struct {
		PipelineID     string               `json:"pipeline_id"`
		DestinationID  string               `json:"destination_id"`
		Reconciliation ReconciliationPolicy `json:"reconciliation"`
		Source         sourceJSON           `json:"source"`
		Links          *LinkState           `json:"links,omitempty"`
	}
	return json.Marshal(ownerJSON{
		PipelineID:     o.Scope.PipelineID,
		DestinationID:  o.Scope.DestinationID,
		Reconciliation: o.Reconciliation,
		Source:         sourceJSON{Manifest: o.Source.Manifest},
		Links:          o.Links,
	})
}

func (o SharedRootOutputFile) MarshalJSON() ([]byte, error) {
	type outputJSON struct {
		Path          string `json:"path"`
		Kind          string `json:"kind"`
		SourcePath    string `json:"source_path"`
		Transform     string `json:"transform,omitempty"`
		URL           string `json:"url,omitempty"`
		SHA256        string `json:"sha256"`
		Size          int64  `json:"size"`
		PipelineID    string `json:"pipeline_id"`
		DestinationID string `json:"destination_id"`
		SourceID      string `json:"source_id"`
		SourceDigest  string `json:"source_digest"`
		SourceCreated string `json:"source_created"`
		CreatedAt     string `json:"created_at"`
		UpdatedAt     string `json:"updated_at"`
	}
	return json.Marshal(outputJSON{
		Path:          o.Path,
		Kind:          o.Kind,
		SourcePath:    o.SourcePath,
		Transform:     o.Transform,
		URL:           o.URL,
		SHA256:        o.SHA256,
		Size:          o.Size,
		PipelineID:    o.Owner.PipelineID,
		DestinationID: o.Owner.DestinationID,
		SourceID:      o.SourceID,
		SourceDigest:  o.SourceDigest,
		SourceCreated: o.SourceCreatedString(),
		CreatedAt:     o.CreatedAtString(),
		UpdatedAt:     o.UpdatedAtString(),
	})
}

func ValidateSharedRoot(s SharedRootState) error {
	if s.SchemaVersion != SharedRootSchemaVersion {
		return fmt.Errorf("state schema_version must be %d", SharedRootSchemaVersion)
	}
	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 != StateModeSharedRoot {
		return fmt.Errorf("state state.mode must be %s", StateModeSharedRoot)
	}
	if s.Owners == nil {
		return fmt.Errorf("state owners is required")
	}
	owners := make(map[OwnerScope]OwnerRecord, len(s.Owners))
	for index, owner := range s.Owners {
		if err := validateOwnerRecord(index, owner); err != nil {
			return err
		}
		if _, exists := owners[owner.Scope]; exists {
			return fmt.Errorf("state owners[%d] duplicates owner %s/%s", index, owner.Scope.PipelineID, owner.Scope.DestinationID)
		}
		owners[owner.Scope] = owner
	}
	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 := validateSharedRootOutput(index, output, owners); 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 validateOwnerRecord(index int, owner OwnerRecord) error {
	if owner.Scope.PipelineID == "" {
		return fmt.Errorf("state owners[%d].pipeline_id is required", index)
	}
	if owner.Scope.DestinationID == "" {
		return fmt.Errorf("state owners[%d].destination_id is required", index)
	}
	if owner.Reconciliation.Mode != config.ReconciliationModeReplace && owner.Reconciliation.Mode != config.ReconciliationModeMerge {
		return fmt.Errorf("state owners[%d].reconciliation.mode must be %s or %s", index, config.ReconciliationModeReplace, config.ReconciliationModeMerge)
	}
	if err := validateEmbeddedManifest(owner.Source.Manifest); err != nil {
		return fmt.Errorf("state owners[%d].source.manifest: %w", index, err)
	}
	if owner.Links != nil && owner.Links.PrimaryURL != "" {
		if err := link.ValidateHTTPURL(owner.Links.PrimaryURL); err != nil {
			return fmt.Errorf("state owners[%d].links.primary_url: %w", index, err)
		}
	}
	return nil
}

func validateSharedRootOutput(index int, output SharedRootOutputFile, owners map[OwnerScope]OwnerRecord) 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 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.Owner.PipelineID == "" {
		return fmt.Errorf("state outputs[%d].pipeline_id is required", index)
	}
	if output.Owner.DestinationID == "" {
		return fmt.Errorf("state outputs[%d].destination_id is required", index)
	}
	if _, exists := owners[output.Owner]; !exists {
		return fmt.Errorf("state outputs[%d] references unknown owner %s/%s", index, output.Owner.PipelineID, output.Owner.DestinationID)
	}
	if output.SourceID == "" {
		return fmt.Errorf("state outputs[%d].source_id is required", index)
	}
	if err := bundle.ValidateDigest(output.SourceDigest); err != nil {
		return fmt.Errorf("state outputs[%d].source_digest: %w", index, err)
	}
	if output.SourceCreated.IsZero() {
		return fmt.Errorf("state outputs[%d].source_created is required", 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
}