notarius/internal/framework/pipeline/references.go

package pipeline

import (
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"math"
	"mime"
	"net/url"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"unicode/utf8"

	"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)

const (
	referenceOriginFile = "file"
	referenceMediaType  = "text/plain"
	unknownMediaType    = "application/octet-stream"
)

type ReferenceMaterializationOptions struct {
	ConfigPath string
	WorkingDir string
}

func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, options ReferenceMaterializationOptions) (ResolvedPipeline, []contracts.ProducerDiagnostic, error) {
	out := resolved
	out.ChunkReferences = CloneReferenceTarget(resolved.ChunkReferences)
	chunkReferenceSet, chunkDiagnostics, err := materializeReferenceTarget(resolved.ID, resolved.ChunkReferences, "", catalog, options)
	if err != nil {
		return ResolvedPipeline{}, nil, err
	}
	out.ChunkReferences.ReferenceSet = chunkReferenceSet
	diagnostics := contracts.CloneProducerDiagnostics(chunkDiagnostics)
	if len(resolved.Steps) == 0 {
		return out, diagnostics, nil
	}
	materializeLane := func(lane ResolvedArtifactLane) (ResolvedArtifactLane, []contracts.ProducerDiagnostic, error) {
		materializedLane := lane
		var allDiagnostics []contracts.ProducerDiagnostic
		materializedLane.ExtractReferences = CloneReferenceTarget(lane.ExtractReferences)
		materializedLane.MergeReferences = CloneReferenceTarget(lane.MergeReferences)
		materializedLane.NormalizeReferences = CloneReferenceTarget(lane.NormalizeReferences)
		extractReferenceSet, laneDiagnostics, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, lane.ArtifactKind, catalog, options)
		if err != nil {
			return ResolvedArtifactLane{}, nil, err
		}
		materializedLane.ExtractReferences.ReferenceSet = extractReferenceSet
		allDiagnostics = append(allDiagnostics, laneDiagnostics...)

		mergeReferenceSet, laneDiagnostics, err := materializeReferenceTarget(resolved.ID, lane.MergeReferences, lane.ArtifactKind, catalog, options)
		if err != nil {
			return ResolvedArtifactLane{}, nil, err
		}
		materializedLane.MergeReferences.ReferenceSet = mergeReferenceSet
		allDiagnostics = append(allDiagnostics, laneDiagnostics...)

		normalizeReferenceSet, laneDiagnostics, err := materializeReferenceTarget(resolved.ID, lane.NormalizeReferences, lane.ArtifactKind, catalog, options)
		if err != nil {
			return ResolvedArtifactLane{}, nil, err
		}
		materializedLane.NormalizeReferences.ReferenceSet = normalizeReferenceSet
		allDiagnostics = append(allDiagnostics, laneDiagnostics...)
		return materializedLane, allDiagnostics, nil
	}
	if len(resolved.Steps) > 0 {
		out.Steps = make([]ResolvedPipelineStep, len(resolved.Steps))
		for i, step := range resolved.Steps {
			out.Steps[i].ID = step.ID
			out.Steps[i].ArtifactLanes = make([]ResolvedArtifactLane, len(step.ArtifactLanes))
			for j, lane := range step.ArtifactLanes {
				materializedLane, laneDiagnostics, err := materializeLane(lane)
				if err != nil {
					return ResolvedPipeline{}, nil, err
				}
				out.Steps[i].ArtifactLanes[j] = materializedLane
				diagnostics = append(diagnostics, laneDiagnostics...)
			}
		}
	}
	return out, diagnostics, nil
}

func materializeReferenceTarget(
	pipelineID string,
	target ResolvedReferenceTarget,
	artifactKind contracts.ArtifactKind,
	catalog ModuleCatalog,
	options ReferenceMaterializationOptions,
) (contracts.ReferenceSet, []contracts.ProducerDiagnostic, error) {
	if len(target.Bindings) == 0 {
		return contracts.ReferenceSet{}, nil, nil
	}
	spec, err := referenceTargetSpec(target, artifactKind, catalog)
	if err != nil {
		return contracts.ReferenceSet{}, nil, fmt.Errorf("%s: %w", referenceTargetContext(pipelineID, target), err)
	}

	slotByName := make(map[string]contracts.ReferenceSlot, len(spec.ReferenceSlots))
	for _, slot := range spec.ReferenceSlots {
		slotByName[slot.Name] = slot
	}

	set := contracts.ReferenceSet{Slots: make(map[string]contracts.ResolvedReferenceSlot, len(target.Bindings))}
	var diagnostics []contracts.ProducerDiagnostic
	for _, binding := range target.Bindings {
		slotName := strings.TrimSpace(binding.SlotName)
		slot, ok := slotByName[slotName]
		if !ok {
			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q is not declared by %s module %q", referenceTargetContext(pipelineID, target), slotName, target.Stage, target.Module)
		}
		if binding.Artifact != nil {
			// Generated selectors are resolved at the step handoff. Keep the
			// declared slot and its constraints, but do not materialize bytes yet.
			set.Slots[slotName] = contracts.ResolvedReferenceSlot{Slot: cloneReferenceSlot(slot)}
			continue
		}

		path, err := referencePath(binding, options)
		if err != nil {
			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q: %w", referenceTargetContext(pipelineID, target), slotName, binding.Source, err)
		}
		file, err := os.Open(path)
		if err != nil {
			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q read %q: %w", referenceTargetContext(pipelineID, target), slotName, path, err)
		}
		regularFileSize := int64(0)
		if info, statErr := file.Stat(); statErr == nil && info.Mode().IsRegular() {
			regularFileSize = info.Size()
		}
		content, err := readReferenceContent(file, slot.MaxBytes)
		closeErr := file.Close()
		if err == nil && closeErr != nil {
			err = closeErr
		}
		if err != nil {
			var sizeErr *referenceSizeLimitError
			if errors.As(err, &sizeErr) {
				sizeBytes := sizeErr.SizeBytes
				if regularFileSize > sizeBytes {
					sizeBytes = regularFileSize
				}
				return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q is %d bytes, limit %d", referenceTargetContext(pipelineID, target), slotName, path, sizeBytes, slot.MaxBytes)
			}
			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q read %q: %w", referenceTargetContext(pipelineID, target), slotName, path, err)
		}
		if !utf8.Valid(content) {
			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q must be UTF-8 text", referenceTargetContext(pipelineID, target), slotName, path)
		}
		mediaType := referenceMediaTypeForPath(path)
		if !referenceMediaTypeAccepted(mediaType, slot.AcceptedMediaTypes) {
			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q media type %q is not accepted", referenceTargetContext(pipelineID, target), slotName, path, mediaType)
		}
		if len(content) == 0 {
			diagnostics = append(diagnostics, contracts.ProducerDiagnostic{Disposition: contracts.DiagnosticDispositionWarning, Category: contracts.DiagnosticCategoryConfiguration, ReasonCode: "empty_reference", OccurrenceCount: 1, Samples: []contracts.DiagnosticSample{{Scope: referenceWarningScope(pipelineID, target, slotName), Message: fmt.Sprintf("reference slot %q for %s is bound to an empty file", slotName, referenceTargetLabel(target))}}})
		}

		item := contracts.ReferenceItem{
			SlotName:      slotName,
			MediaType:     mediaType,
			Content:       append([]byte(nil), content...),
			Digest:        referenceDigest(content),
			Origin:        contracts.ReferenceOrigin{Type: referenceOriginFile, URI: fileURI(path)},
			SizeBytes:     int64(len(content)),
			BindingSource: strings.TrimSpace(binding.BindingSource),
		}
		set.Slots[slotName] = contracts.ResolvedReferenceSlot{
			Slot:  cloneReferenceSlot(slot),
			Items: []contracts.ReferenceItem{item},
		}
	}
	return set, diagnostics, nil
}

type referenceSizeLimitError struct {
	SizeBytes int64
	MaxBytes  int64
}

func (err *referenceSizeLimitError) Error() string {
	return fmt.Sprintf("reference is %d bytes, limit %d", err.SizeBytes, err.MaxBytes)
}

func readReferenceContent(reader io.Reader, maxBytes int64) ([]byte, error) {
	if maxBytes <= 0 {
		return io.ReadAll(reader)
	}
	if maxBytes == math.MaxInt64 {
		return nil, fmt.Errorf("maximum reference size %d cannot be safely bounded", maxBytes)
	}
	content, err := io.ReadAll(io.LimitReader(reader, maxBytes+1))
	if err != nil {
		return nil, err
	}
	if int64(len(content)) > maxBytes {
		return nil, &referenceSizeLimitError{SizeBytes: int64(len(content)), MaxBytes: maxBytes}
	}
	return content, nil
}

func referenceTargetSpec(target ResolvedReferenceTarget, artifactKind contracts.ArtifactKind, catalog ModuleCatalog) (ModuleSpec, error) {
	switch target.Stage {
	case StageChunk:
		return registrySpec(catalog.Chunkers, target.Module)
	case StageExtract:
		return registrySpec(catalog.Extractors, target.Module)
	case StageMerge:
		if catalog.Mergers == nil {
			return ModuleSpec{}, fmt.Errorf("module %q is not registered", target.Module)
		}
		if spec, ok := catalog.Mergers.SpecForArtifact(target.Module, artifactKind); ok {
			return spec, nil
		}
		return ModuleSpec{}, missingArtifactVariantError("merger", target.Module, artifactKind, catalog.Mergers.registeredKinds(target.Module))
	case StageNormalize:
		if catalog.Normalizers == nil {
			return ModuleSpec{}, fmt.Errorf("module %q is not registered", target.Module)
		}
		if spec, ok := catalog.Normalizers.SpecForArtifact(target.Module, artifactKind); ok {
			return spec, nil
		}
		return ModuleSpec{}, missingArtifactVariantError("normalizer", target.Module, artifactKind, catalog.Normalizers.registeredKinds(target.Module))
	default:
		return ModuleSpec{}, fmt.Errorf("reference target stage %q is not supported", target.Stage)
	}
}

func referenceTargetContext(pipelineID string, target ResolvedReferenceTarget) string {
	if target.LaneID != "" {
		return fmt.Sprintf("pipeline %q lane %q %s module %q", pipelineID, target.LaneID, target.Stage, target.Module)
	}
	return fmt.Sprintf("pipeline %q %s module %q", pipelineID, target.Stage, target.Module)
}

func referenceTargetLabel(target ResolvedReferenceTarget) string {
	if target.LaneID != "" {
		return fmt.Sprintf("lane %q %s target", target.LaneID, target.Stage)
	}
	return fmt.Sprintf("%s target", target.Stage)
}

func referenceWarningScope(pipelineID string, target ResolvedReferenceTarget, slotName string) string {
	if target.LaneID != "" {
		return fmt.Sprintf("pipeline.%s.lane.%s.%s.reference.%s", pipelineID, target.LaneID, target.Stage, slotName)
	}
	return fmt.Sprintf("pipeline.%s.%s.reference.%s", pipelineID, target.Stage, slotName)
}

func referenceMediaTypeForPath(path string) string {
	extension := strings.ToLower(filepath.Ext(path))
	mediaType := mime.TypeByExtension(extension)
	if strings.TrimSpace(mediaType) == "" {
		if extension == ".md" || extension == ".markdown" {
			return "text/markdown"
		}
		if extension == ".yaml" || extension == ".yml" {
			return "application/yaml"
		}
		return unknownMediaType
	}
	return canonicalMediaType(mediaType)
}

func referenceMediaTypeAccepted(mediaType string, accepted []string) bool {
	if len(accepted) == 0 {
		return true
	}
	mediaType = canonicalMediaType(mediaType)
	for _, value := range accepted {
		if strings.EqualFold(mediaType, canonicalMediaType(value)) {
			return true
		}
	}
	return false
}

func canonicalMediaType(mediaType string) string {
	trimmed := strings.TrimSpace(mediaType)
	if trimmed == "" {
		return ""
	}
	parsed, _, err := mime.ParseMediaType(trimmed)
	if err != nil {
		return strings.ToLower(trimmed)
	}
	return strings.ToLower(parsed)
}

func referencePath(binding ReferenceBinding, options ReferenceMaterializationOptions) (string, error) {
	source := strings.TrimSpace(binding.Source)
	if source == "" {
		return "", fmt.Errorf("must not be empty")
	}
	if filepath.IsAbs(source) {
		return filepath.Clean(source), nil
	}

	base := strings.TrimSpace(options.WorkingDir)
	if strings.TrimSpace(binding.BindingSource) == contracts.ReferenceBindingSourceConfig {
		base = filepath.Dir(strings.TrimSpace(options.ConfigPath))
	}
	if base == "" {
		var err error
		base, err = os.Getwd()
		if err != nil {
			return "", fmt.Errorf("resolve working directory: %w", err)
		}
	}
	return filepath.Clean(filepath.Join(base, source)), nil
}

func referenceDigest(content []byte) string {
	sum := sha256.Sum256(content)
	return "sha256:" + hex.EncodeToString(sum[:])
}

func fileURI(path string) string {
	absolute, err := filepath.Abs(path)
	if err != nil {
		absolute = path
	}
	absolute = filepath.ToSlash(filepath.Clean(absolute))
	if strings.HasPrefix(absolute, "/") {
		return "file://" + (&url.URL{Path: absolute}).EscapedPath()
	}
	return "file:///" + (&url.URL{Path: absolute}).EscapedPath()
}

func cloneReferenceSlot(slot contracts.ReferenceSlot) contracts.ReferenceSlot {
	slot.AcceptedMediaTypes = append([]string(nil), slot.AcceptedMediaTypes...)
	slot.AcceptedArtifactKinds = append([]contracts.ArtifactKind(nil), slot.AcceptedArtifactKinds...)
	return slot
}

func CloneReferenceSet(in contracts.ReferenceSet) contracts.ReferenceSet {
	if len(in.Slots) == 0 {
		return contracts.ReferenceSet{}
	}
	out := contracts.ReferenceSet{Slots: make(map[string]contracts.ResolvedReferenceSlot, len(in.Slots))}
	keys := make([]string, 0, len(in.Slots))
	for key := range in.Slots {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	for _, key := range keys {
		slot := in.Slots[key]
		slot.Slot = cloneReferenceSlot(slot.Slot)
		if len(slot.Items) > 0 {
			items := make([]contracts.ReferenceItem, len(slot.Items))
			for i, item := range slot.Items {
				items[i] = contracts.CloneReferenceItem(item)
			}
			slot.Items = items
		}
		out.Slots[key] = slot
	}
	return out
}

func CloneReferenceTarget(in ResolvedReferenceTarget) ResolvedReferenceTarget {
	out := in
	if len(in.Bindings) > 0 {
		out.Bindings = make([]ReferenceBinding, len(in.Bindings))
		for i, binding := range in.Bindings {
			out.Bindings[i] = binding
			if binding.Artifact != nil {
				artifact := *binding.Artifact
				out.Bindings[i].Artifact = &artifact
			}
		}
	}
	out.ReferenceSet = CloneReferenceSet(in.ReferenceSet)
	return out
}

func ReferenceProvenance(resolved ResolvedPipeline) []artifacts.ReferenceProvenance {
	provenance := []artifacts.ReferenceProvenance{}
	provenance = append(provenance, referenceTargetProvenance(resolved.ChunkReferences)...)
	if len(resolved.Steps) > 0 {
		for _, step := range resolved.Steps {
			for _, lane := range step.ArtifactLanes {
				provenance = append(provenance, referenceTargetProvenance(lane.ExtractReferences)...)
				provenance = append(provenance, referenceTargetProvenance(lane.MergeReferences)...)
				provenance = append(provenance, referenceTargetProvenance(lane.NormalizeReferences)...)
			}
		}
	}
	return provenance
}

func referenceTargetProvenance(target ResolvedReferenceTarget) []artifacts.ReferenceProvenance {
	if len(target.ReferenceSet.Slots) == 0 {
		return nil
	}
	provenance := []artifacts.ReferenceProvenance{}
	slotNames := make([]string, 0, len(target.ReferenceSet.Slots))
	for slotName := range target.ReferenceSet.Slots {
		slotNames = append(slotNames, slotName)
	}
	sort.Strings(slotNames)
	for _, slotName := range slotNames {
		slot := target.ReferenceSet.Slots[slotName]
		for _, item := range slot.Items {
			schemaDigest := ""
			if item.ArtifactSchema.ID != "" || item.ArtifactSchema.Name != "" || item.ArtifactSchema.Version != "" || len(item.ArtifactSchema.JSONSchema) > 0 {
				schemaDigest = contracts.DigestArtifactSchema(item.ArtifactSchema)
			}
			provenance = append(provenance, artifacts.ReferenceProvenance{
				Stage:            string(target.Stage),
				StepID:           target.StepID,
				LaneID:           target.LaneID,
				SlotName:         item.SlotName,
				OriginType:       item.Origin.Type,
				OriginURI:        item.Origin.URI,
				Digest:           item.Digest,
				MediaType:        item.MediaType,
				SizeBytes:        item.SizeBytes,
				BindingSource:    item.BindingSource,
				ArtifactKind:     string(item.ArtifactKind),
				SchemaID:         item.ArtifactSchema.ID,
				SchemaName:       item.ArtifactSchema.Name,
				SchemaVersion:    item.ArtifactSchema.Version,
				SchemaDigest:     schemaDigest,
				ProducerPipeline: item.Producer.PipelineID,
				ProducerStep:     item.Producer.StepID,
				ProducerLane:     item.Producer.LaneID,
				ProducerModule:   item.Producer.ModuleKey,
			})
		}
	}
	return provenance
}