Add presence-aware configuration composition

2026-08-30 12:21:27 +00:00
parent 3b5e9db41f
commit 49ea747b17
3 changed files with 1014 additions and 1 deletions
--- a/internal/config/composition.go
+++ b/internal/config/composition.go
@@ -0,0 +1,657 @@
+package config
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"sort"
+	"strconv"
+	"strings"
+
+	"gopkg.in/yaml.v3"
+)
+
+// compositionDocument is the presence-aware representation used while
+// assembling pipeline configuration sources. It deliberately models YAML
+// mechanics rather than duplicating PipelineConfig's field schema.
+type compositionDocument struct {
+	root    *compositionNode
+	sources []string
+}
+
+type compositionNode struct {
+	kind    yaml.Kind
+	tag     string
+	value   string
+	path    string
+	sources []string
+	line    int
+	column  int
+	fields  []compositionField
+	items   []*compositionNode
+}
+
+type compositionField struct {
+	key    string
+	value  *compositionNode
+	order  int
+	line   int
+	column int
+}
+
+// compositionValueRecord is a deterministic semantic leaf projection. Lists
+// are atomic configuration values, while mappings are traversed recursively.
+type compositionValueRecord struct {
+	Path    string
+	Kind    yaml.Kind
+	Value   string
+	Sources []string
+}
+
+// parseCompositionDocument parses exactly one YAML mapping while retaining
+// source ownership, explicit zero values, and declaration order.
+func parseCompositionDocument(source string, reader io.Reader) (*compositionDocument, error) {
+	if strings.TrimSpace(source) == "" {
+		return nil, fmt.Errorf("configuration source identity is required")
+	}
+	if reader == nil {
+		return nil, fmt.Errorf("configuration source %q: reader is nil", source)
+	}
+
+	decoder := yaml.NewDecoder(reader)
+	var document yaml.Node
+	if err := decoder.Decode(&document); err != nil {
+		if err == io.EOF {
+			return nil, fmt.Errorf("configuration source %q: document is empty", source)
+		}
+		return nil, fmt.Errorf("configuration source %q: decode YAML: %w", source, err)
+	}
+	var trailing yaml.Node
+	if err := decoder.Decode(&trailing); err == nil {
+		return nil, fmt.Errorf("configuration source %q: must contain exactly one YAML document", source)
+	} else if err != io.EOF {
+		return nil, fmt.Errorf("configuration source %q: decode trailing YAML: %w", source, err)
+	}
+	if document.Kind != yaml.DocumentNode || len(document.Content) != 1 {
+		return nil, fmt.Errorf("configuration source %q: must contain exactly one YAML document", source)
+	}
+	if document.Content[0].Kind != yaml.MappingNode {
+		return nil, fmt.Errorf(
+			"configuration source %q: top-level document must be a mapping, got %s",
+			source,
+			yamlKindName(document.Content[0].Kind),
+		)
+	}
+
+	root, err := buildCompositionNode(document.Content[0], source, "")
+	if err != nil {
+		return nil, err
+	}
+	return &compositionDocument{root: root, sources: []string{source}}, nil
+}
+
+func parseCompositionBytes(source string, data []byte) (*compositionDocument, error) {
+	return parseCompositionDocument(source, bytes.NewReader(data))
+}
+
+func buildCompositionNode(node *yaml.Node, source, path string) (*compositionNode, error) {
+	if node == nil {
+		return nil, fmt.Errorf("configuration source %q at %s: YAML node is nil", source, displayCompositionPath(path))
+	}
+	if node.Kind == yaml.AliasNode {
+		return nil, compositionNodeError(source, path, node, "YAML aliases are not supported because source ownership would be ambiguous")
+	}
+	result := &compositionNode{
+		kind: node.Kind, tag: node.Tag, value: node.Value, path: path,
+		sources: []string{source}, line: node.Line, column: node.Column,
+	}
+
+	switch node.Kind {
+	case yaml.MappingNode:
+		if len(node.Content)%2 != 0 {
+			return nil, compositionNodeError(source, path, node, "mapping has an incomplete key/value pair")
+		}
+		seen := make(map[string]*yaml.Node, len(node.Content)/2)
+		for index := 0; index < len(node.Content); index += 2 {
+			keyNode := node.Content[index]
+			valueNode := node.Content[index+1]
+			if keyNode.Kind == yaml.AliasNode {
+				return nil, compositionNodeError(source, path, keyNode, "YAML aliases are not supported because source ownership would be ambiguous")
+			}
+			if valueNode.Kind == yaml.AliasNode {
+				return nil, compositionNodeError(source, appendCompositionPath(path, keyNode.Value), valueNode, "YAML aliases are not supported because source ownership would be ambiguous")
+			}
+			if keyNode.Kind != yaml.ScalarNode || keyNode.Tag != "!!str" {
+				return nil, compositionNodeError(source, path, keyNode, "mapping keys must be strings")
+			}
+			key := keyNode.Value
+			fieldPath := appendCompositionPath(path, key)
+			if prior, duplicate := seen[key]; duplicate {
+				return nil, fmt.Errorf(
+					"configuration source %q at %s: duplicate YAML key %q (first declared at line %d, column %d; repeated at line %d, column %d)",
+					source, displayCompositionPath(fieldPath), key,
+					prior.Line, prior.Column, keyNode.Line, keyNode.Column,
+				)
+			}
+			seen[key] = keyNode
+			child, err := buildCompositionNode(valueNode, source, fieldPath)
+			if err != nil {
+				return nil, err
+			}
+			result.fields = append(result.fields, compositionField{
+				key: key, value: child, order: len(result.fields),
+				line: keyNode.Line, column: keyNode.Column,
+			})
+		}
+	case yaml.SequenceNode:
+		for index, childNode := range node.Content {
+			childPath := fmt.Sprintf("%s[%d]", path, index)
+			child, err := buildCompositionNode(childNode, source, childPath)
+			if err != nil {
+				return nil, err
+			}
+			result.items = append(result.items, child)
+		}
+	case yaml.ScalarNode:
+		// Scalar tag and lexical value retain distinctions such as explicit
+		// false, zero, an empty string, and null until final strict decoding.
+	default:
+		return nil, compositionNodeError(source, path, node, fmt.Sprintf("unsupported YAML node kind %s", yamlKindName(node.Kind)))
+	}
+	return result, nil
+}
+
+// mergeAdditiveComposition recursively combines disjoint mappings. A scalar,
+// list, or final keyed value may have only one base owner, regardless of
+// whether duplicate values happen to be equal.
+func mergeAdditiveComposition(base, incoming *compositionDocument) (*compositionDocument, error) {
+	if err := validateCompositionDocument(base, "base"); err != nil {
+		return nil, err
+	}
+	if err := validateCompositionDocument(incoming, "incoming"); err != nil {
+		return nil, err
+	}
+	merged, err := mergeAdditiveNodes(cloneCompositionNode(base.root), incoming.root)
+	if err != nil {
+		return nil, err
+	}
+	return &compositionDocument{
+		root: merged,
+		sources: appendUniqueStrings(
+			append([]string(nil), base.sources...), incoming.sources...,
+		),
+	}, nil
+}
+
+func mergeAdditiveNodes(base, incoming *compositionNode) (*compositionNode, error) {
+	if base.kind != yaml.MappingNode || incoming.kind != yaml.MappingNode {
+		return nil, newCompositionConflict("additive base merge", base.path, base, incoming)
+	}
+	base.sources = appendUniqueStrings(base.sources, incoming.sources...)
+	for _, incomingField := range incoming.fields {
+		index := compositionFieldIndex(base.fields, incomingField.key)
+		if index < 0 {
+			field := cloneCompositionField(incomingField)
+			field.order = len(base.fields)
+			base.fields = append(base.fields, field)
+			continue
+		}
+		baseValue := base.fields[index].value
+		incomingValue := incomingField.value
+		if baseValue.kind == yaml.MappingNode && incomingValue.kind == yaml.MappingNode {
+			if len(baseValue.fields) == 0 || len(incomingValue.fields) == 0 {
+				return nil, newCompositionConflict("additive base merge", incomingValue.path, baseValue, incomingValue)
+			}
+			merged, err := mergeAdditiveNodes(baseValue, incomingValue)
+			if err != nil {
+				return nil, err
+			}
+			base.fields[index].value = merged
+			continue
+		}
+		return nil, newCompositionConflict("additive base merge", incomingValue.path, baseValue, incomingValue)
+	}
+	return base, nil
+}
+
+// mergeOverlayComposition applies the sole overwrite layer. Mappings merge
+// recursively; same-kind scalars and lists replace; null and kind changes are
+// rejected.
+func mergeOverlayComposition(base, overlay *compositionDocument) (*compositionDocument, error) {
+	if err := validateCompositionDocument(base, "base"); err != nil {
+		return nil, err
+	}
+	if err := validateCompositionDocument(overlay, "overlay"); err != nil {
+		return nil, err
+	}
+	if null := firstNullCompositionNode(overlay.root); null != nil {
+		return nil, fmt.Errorf(
+			"configuration overlay at %s from %s: null cannot delete an effective value",
+			displayCompositionPath(null.path), formatCompositionSources(null.sources),
+		)
+	}
+	merged, err := mergeOverlayNodes(cloneCompositionNode(base.root), overlay.root)
+	if err != nil {
+		return nil, err
+	}
+	return &compositionDocument{
+		root: merged,
+		sources: appendUniqueStrings(
+			append([]string(nil), base.sources...), overlay.sources...,
+		),
+	}, nil
+}
+
+func mergeOverlayNodes(base, overlay *compositionNode) (*compositionNode, error) {
+	if base.kind != yaml.MappingNode || overlay.kind != yaml.MappingNode {
+		return nil, newCompositionConflict("profile overlay", base.path, base, overlay)
+	}
+	base.sources = appendUniqueStrings(base.sources, overlay.sources...)
+	for _, overlayField := range overlay.fields {
+		index := compositionFieldIndex(base.fields, overlayField.key)
+		if index < 0 {
+			field := cloneCompositionField(overlayField)
+			field.order = len(base.fields)
+			base.fields = append(base.fields, field)
+			continue
+		}
+		baseValue := base.fields[index].value
+		overlayValue := overlayField.value
+		if baseValue.kind != overlayValue.kind {
+			return nil, newCompositionConflict("profile overlay kind change", overlayValue.path, baseValue, overlayValue)
+		}
+		if baseValue.kind == yaml.MappingNode {
+			merged, err := mergeOverlayNodes(baseValue, overlayValue)
+			if err != nil {
+				return nil, err
+			}
+			base.fields[index].value = merged
+			continue
+		}
+		base.fields[index].value = cloneCompositionNode(overlayValue)
+	}
+	return base, nil
+}
+
+// canonicalYAML renders the effective mapping with sorted keys and normalized
+// presentation while retaining sequence order and scalar YAML types.
+func (document *compositionDocument) canonicalYAML() ([]byte, error) {
+	if err := validateCompositionDocument(document, "document"); err != nil {
+		return nil, err
+	}
+	root, err := compositionYAMLNode(document.root)
+	if err != nil {
+		return nil, err
+	}
+	var buffer bytes.Buffer
+	encoder := yaml.NewEncoder(&buffer)
+	encoder.SetIndent(2)
+	if err := encoder.Encode(root); err != nil {
+		return nil, fmt.Errorf("render effective configuration YAML: %w", err)
+	}
+	if err := encoder.Close(); err != nil {
+		return nil, fmt.Errorf("finish effective configuration YAML: %w", err)
+	}
+	return buffer.Bytes(), nil
+}
+
+// canonicalDigestInput provides a deterministic, formatting-independent byte
+// representation for later secret-free effective configuration digesting.
+func (document *compositionDocument) canonicalDigestInput() ([]byte, error) {
+	if err := validateCompositionDocument(document, "document"); err != nil {
+		return nil, err
+	}
+	value, err := canonicalCompositionValue(document.root)
+	if err != nil {
+		return nil, err
+	}
+	data, err := json.Marshal(value)
+	if err != nil {
+		return nil, fmt.Errorf("serialize canonical configuration digest input: %w", err)
+	}
+	return append(data, '\n'), nil
+}
+
+// semanticRecords returns sorted atomic values for future effective diff and
+// source-report projections. A caller receives copies of all source slices.
+func (document *compositionDocument) semanticRecords() ([]compositionValueRecord, error) {
+	if err := validateCompositionDocument(document, "document"); err != nil {
+		return nil, err
+	}
+	var records []compositionValueRecord
+	if err := appendCompositionRecords(document.root, &records); err != nil {
+		return nil, err
+	}
+	sort.Slice(records, func(i, j int) bool { return records[i].Path < records[j].Path })
+	return records, nil
+}
+
+func appendCompositionRecords(node *compositionNode, records *[]compositionValueRecord) error {
+	if node.kind == yaml.MappingNode && len(node.fields) > 0 {
+		for _, field := range node.fields {
+			if err := appendCompositionRecords(field.value, records); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+	value, err := canonicalCompositionValue(node)
+	if err != nil {
+		return err
+	}
+	encoded, err := json.Marshal(value)
+	if err != nil {
+		return fmt.Errorf("serialize configuration value at %s: %w", displayCompositionPath(node.path), err)
+	}
+	*records = append(*records, compositionValueRecord{
+		Path: node.path, Kind: node.kind, Value: string(encoded),
+		Sources: append([]string(nil), node.sources...),
+	})
+	return nil
+}
+
+type canonicalCompositionField struct {
+	Key   string `json:"key"`
+	Value any    `json:"value"`
+}
+
+type canonicalCompositionNode struct {
+	Kind   string                      `json:"kind"`
+	Tag    string                      `json:"tag,omitempty"`
+	Value  any                         `json:"value,omitempty"`
+	Fields []canonicalCompositionField `json:"fields,omitempty"`
+	Items  []any                       `json:"items,omitempty"`
+}
+
+func canonicalCompositionValue(node *compositionNode) (any, error) {
+	switch node.kind {
+	case yaml.MappingNode:
+		fields := append([]compositionField(nil), node.fields...)
+		sort.Slice(fields, func(i, j int) bool { return fields[i].key < fields[j].key })
+		result := canonicalCompositionNode{Kind: "mapping"}
+		if len(fields) == 0 {
+			result.Fields = []canonicalCompositionField{}
+		}
+		for _, field := range fields {
+			value, err := canonicalCompositionValue(field.value)
+			if err != nil {
+				return nil, err
+			}
+			result.Fields = append(result.Fields, canonicalCompositionField{Key: field.key, Value: value})
+		}
+		return result, nil
+	case yaml.SequenceNode:
+		result := canonicalCompositionNode{Kind: "sequence", Items: make([]any, 0, len(node.items))}
+		for _, item := range node.items {
+			value, err := canonicalCompositionValue(item)
+			if err != nil {
+				return nil, err
+			}
+			result.Items = append(result.Items, value)
+		}
+		return result, nil
+	case yaml.ScalarNode:
+		value, err := canonicalScalarValue(node)
+		if err != nil {
+			return nil, err
+		}
+		return canonicalCompositionNode{Kind: "scalar", Tag: node.tag, Value: value}, nil
+	default:
+		return nil, fmt.Errorf("configuration at %s has unsupported YAML kind %s", displayCompositionPath(node.path), yamlKindName(node.kind))
+	}
+}
+
+func canonicalScalarValue(node *compositionNode) (any, error) {
+	raw := &yaml.Node{Kind: yaml.ScalarNode, Tag: node.tag, Value: node.value}
+	var value any
+	if err := raw.Decode(&value); err != nil {
+		return nil, fmt.Errorf("decode scalar at %s: %w", displayCompositionPath(node.path), err)
+	}
+	switch typed := value.(type) {
+	case nil, bool, string, int, int64, uint64, float64:
+		return typed, nil
+	default:
+		// yaml.v3 may decode timestamps or uncommon scalar tags into types that
+		// encoding/json can serialize deterministically. Preserve the resolved
+		// tag alongside the value in the containing canonical node.
+		return typed, nil
+	}
+}
+
+func compositionYAMLNode(node *compositionNode) (*yaml.Node, error) {
+	switch node.kind {
+	case yaml.MappingNode:
+		result := &yaml.Node{Kind: yaml.MappingNode, Tag: "!!map"}
+		fields := append([]compositionField(nil), node.fields...)
+		sort.Slice(fields, func(i, j int) bool { return fields[i].key < fields[j].key })
+		for _, field := range fields {
+			value, err := compositionYAMLNode(field.value)
+			if err != nil {
+				return nil, err
+			}
+			result.Content = append(result.Content,
+				&yaml.Node{Kind: yaml.ScalarNode, Tag: "!!str", Value: field.key},
+				value,
+			)
+		}
+		return result, nil
+	case yaml.SequenceNode:
+		result := &yaml.Node{Kind: yaml.SequenceNode, Tag: "!!seq"}
+		for _, item := range node.items {
+			value, err := compositionYAMLNode(item)
+			if err != nil {
+				return nil, err
+			}
+			result.Content = append(result.Content, value)
+		}
+		return result, nil
+	case yaml.ScalarNode:
+		return normalizedCompositionScalarNode(node)
+	default:
+		return nil, fmt.Errorf("configuration at %s has unsupported YAML kind %s", displayCompositionPath(node.path), yamlKindName(node.kind))
+	}
+}
+
+func normalizedCompositionScalarNode(node *compositionNode) (*yaml.Node, error) {
+	raw := &yaml.Node{Kind: yaml.ScalarNode, Tag: node.tag, Value: node.value}
+	var value any
+	if err := raw.Decode(&value); err != nil {
+		return nil, fmt.Errorf("normalize scalar at %s: %w", displayCompositionPath(node.path), err)
+	}
+	normalized := &yaml.Node{}
+	if err := normalized.Encode(value); err != nil {
+		return nil, fmt.Errorf("encode normalized scalar at %s: %w", displayCompositionPath(node.path), err)
+	}
+	if normalized.Kind != yaml.ScalarNode {
+		return nil, fmt.Errorf("normalize scalar at %s produced YAML kind %s", displayCompositionPath(node.path), yamlKindName(normalized.Kind))
+	}
+	return normalized, nil
+}
+
+func validateCompositionDocument(document *compositionDocument, role string) error {
+	if document == nil || document.root == nil {
+		return fmt.Errorf("configuration composition %s document is nil", role)
+	}
+	if document.root.kind != yaml.MappingNode {
+		return fmt.Errorf("configuration composition %s root must be a mapping", role)
+	}
+	return nil
+}
+
+func firstNullCompositionNode(node *compositionNode) *compositionNode {
+	if node == nil {
+		return nil
+	}
+	if node.kind == yaml.ScalarNode && node.tag == "!!null" {
+		return node
+	}
+	for _, field := range node.fields {
+		if found := firstNullCompositionNode(field.value); found != nil {
+			return found
+		}
+	}
+	for _, item := range node.items {
+		if found := firstNullCompositionNode(item); found != nil {
+			return found
+		}
+	}
+	return nil
+}
+
+func cloneCompositionNode(node *compositionNode) *compositionNode {
+	if node == nil {
+		return nil
+	}
+	clone := &compositionNode{
+		kind: node.kind, tag: node.tag, value: node.value, path: node.path,
+		sources: append([]string(nil), node.sources...), line: node.line, column: node.column,
+	}
+	for _, field := range node.fields {
+		clone.fields = append(clone.fields, cloneCompositionField(field))
+	}
+	for _, item := range node.items {
+		clone.items = append(clone.items, cloneCompositionNode(item))
+	}
+	return clone
+}
+
+func cloneCompositionField(field compositionField) compositionField {
+	return compositionField{
+		key: field.key, value: cloneCompositionNode(field.value), order: field.order,
+		line: field.line, column: field.column,
+	}
+}
+
+func compositionFieldIndex(fields []compositionField, key string) int {
+	for index := range fields {
+		if fields[index].key == key {
+			return index
+		}
+	}
+	return -1
+}
+
+func appendCompositionPath(parent, key string) string {
+	if isSimpleCompositionPathSegment(key) {
+		if parent == "" {
+			return key
+		}
+		return parent + "." + key
+	}
+	if parent == "" {
+		return "[" + strconv.Quote(key) + "]"
+	}
+	return parent + "[" + strconv.Quote(key) + "]"
+}
+
+func isSimpleCompositionPathSegment(value string) bool {
+	if value == "" {
+		return false
+	}
+	for index, char := range value {
+		if (char >= 'a' && char <= 'z') || (char >= 'A' && char <= 'Z') || char == '_' || (index > 0 && char >= '0' && char <= '9') || (index > 0 && char == '-') {
+			continue
+		}
+		return false
+	}
+	return true
+}
+
+func displayCompositionPath(path string) string {
+	if path == "" {
+		return "<root>"
+	}
+	return path
+}
+
+func compositionNodeError(source, path string, node *yaml.Node, message string) error {
+	line, column := 0, 0
+	if node != nil {
+		line, column = node.Line, node.Column
+	}
+	return fmt.Errorf(
+		"configuration source %q at %s (line %d, column %d): %s",
+		source, displayCompositionPath(path), line, column, message,
+	)
+}
+
+func newCompositionConflict(operation, path string, values ...*compositionNode) error {
+	var sources []string
+	var kinds []string
+	for _, value := range values {
+		if value == nil {
+			continue
+		}
+		sources = appendUniqueStrings(sources, compositionClaimSources(value)...)
+		kind := yamlKindName(value.kind)
+		if !containsString(kinds, kind) {
+			kinds = append(kinds, kind)
+		}
+	}
+	return fmt.Errorf(
+		"configuration %s conflict at %s: claimed by %s (YAML kinds: %s)",
+		operation, displayCompositionPath(path), formatCompositionSources(sources), strings.Join(kinds, ", "),
+	)
+}
+
+func compositionClaimSources(node *compositionNode) []string {
+	if node == nil {
+		return nil
+	}
+	sources := append([]string(nil), node.sources...)
+	for _, field := range node.fields {
+		sources = appendUniqueStrings(sources, compositionClaimSources(field.value)...)
+	}
+	for _, item := range node.items {
+		sources = appendUniqueStrings(sources, compositionClaimSources(item)...)
+	}
+	return sources
+}
+
+func appendUniqueStrings(values []string, additions ...string) []string {
+	seen := make(map[string]struct{}, len(values)+len(additions))
+	result := make([]string, 0, len(values)+len(additions))
+	for _, value := range append(append([]string(nil), values...), additions...) {
+		if _, exists := seen[value]; exists {
+			continue
+		}
+		seen[value] = struct{}{}
+		result = append(result, value)
+	}
+	return result
+}
+
+func containsString(values []string, target string) bool {
+	for _, value := range values {
+		if value == target {
+			return true
+		}
+	}
+	return false
+}
+
+func formatCompositionSources(sources []string) string {
+	quoted := make([]string, 0, len(sources))
+	for _, source := range sources {
+		quoted = append(quoted, strconv.Quote(source))
+	}
+	return strings.Join(quoted, ", ")
+}
+
+func yamlKindName(kind yaml.Kind) string {
+	switch kind {
+	case yaml.DocumentNode:
+		return "document"
+	case yaml.MappingNode:
+		return "mapping"
+	case yaml.SequenceNode:
+		return "sequence"
+	case yaml.ScalarNode:
+		return "scalar"
+	case yaml.AliasNode:
+		return "alias"
+	default:
+		return fmt.Sprintf("kind(%d)", kind)
+	}
+}