distributor/internal/state/catalog.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 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
}