notarius/internal/framework/pipeline/references.go

package pipeline

import (
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"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.Warning, error) {
	out := resolved
	out.ChunkReferences = CloneReferenceTarget(resolved.ChunkReferences)
	chunkReferenceSet, chunkWarnings, err := materializeReferenceTarget(resolved.ID, resolved.ChunkReferences, catalog, options)
	if err != nil {
		return ResolvedPipeline{}, nil, err
	}
	out.ChunkReferences.ReferenceSet = chunkReferenceSet
	warnings := append([]contracts.Warning(nil), chunkWarnings...)
	if len(resolved.ArtifactLanes) == 0 {
		return out, warnings, nil
	}

	out.ArtifactLanes = make([]ResolvedArtifactLane, len(resolved.ArtifactLanes))
	for i, lane := range resolved.ArtifactLanes {
		materializedLane := lane
		materializedLane.ExtractReferences = CloneReferenceTarget(lane.ExtractReferences)
		materializedLane.MergeReferences = CloneReferenceTarget(lane.MergeReferences)
		materializedLane.NormalizeReferences = CloneReferenceTarget(lane.NormalizeReferences)
		extractReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, catalog, options)
		if err != nil {
			return ResolvedPipeline{}, nil, err
		}
		materializedLane.ExtractReferences.ReferenceSet = extractReferenceSet
		warnings = append(warnings, laneWarnings...)

		mergeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.MergeReferences, catalog, options)
		if err != nil {
			return ResolvedPipeline{}, nil, err
		}
		materializedLane.MergeReferences.ReferenceSet = mergeReferenceSet
		warnings = append(warnings, laneWarnings...)

		normalizeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.NormalizeReferences, catalog, options)
		if err != nil {
			return ResolvedPipeline{}, nil, err
		}
		materializedLane.NormalizeReferences.ReferenceSet = normalizeReferenceSet
		warnings = append(warnings, laneWarnings...)
		out.ArtifactLanes[i] = materializedLane
	}
	return out, warnings, nil
}

func materializeReferenceTarget(
	pipelineID string,
	target ResolvedReferenceTarget,
	catalog ModuleCatalog,
	options ReferenceMaterializationOptions,
) (contracts.ReferenceSet, []contracts.Warning, error) {
	if len(target.Bindings) == 0 {
		return contracts.ReferenceSet{}, nil, nil
	}
	spec, err := referenceTargetSpec(target, 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 warnings []contracts.Warning
	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)
		}

		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)
		}
		content, err := os.ReadFile(path)
		if err != nil {
			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 slot.MaxBytes > 0 && int64(len(content)) > slot.MaxBytes {
			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q is %d bytes, limit %d", referenceTargetContext(pipelineID, target), slotName, path, len(content), slot.MaxBytes)
		}
		if len(content) == 0 {
			warnings = append(warnings, contracts.Warning{
				Scope:      referenceWarningScope(pipelineID, target, slotName),
				ReasonCode: "empty_reference",
				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, warnings, nil
}

func referenceTargetSpec(target ResolvedReferenceTarget, 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:
		return registrySpec(catalog.Mergers, target.Module)
	case StageNormalize:
		return registrySpec(catalog.Normalizers, 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...)
	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 {
				item.Content = append([]byte(nil), item.Content...)
				items[i] = item
			}
			slot.Items = items
		}
		out.Slots[key] = slot
	}
	return out
}

func CloneReferenceTarget(in ResolvedReferenceTarget) ResolvedReferenceTarget {
	out := in
	out.Bindings = append([]ReferenceBinding(nil), in.Bindings...)
	out.ReferenceSet = CloneReferenceSet(in.ReferenceSet)
	return out
}

func ReferenceProvenance(resolved ResolvedPipeline) []artifacts.ReferenceProvenance {
	provenance := []artifacts.ReferenceProvenance{}
	provenance = append(provenance, referenceTargetProvenance(resolved.ChunkReferences)...)
	for _, lane := range resolved.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 {
			provenance = append(provenance, artifacts.ReferenceProvenance{
				Stage:         string(target.Stage),
				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,
			})
		}
	}
	return provenance
}