ConfigKit/internal/engine/import_acquisition.go

package engine

import (
	"bytes"
	"errors"
	"io/fs"
	"strconv"

	"gitea.maximumdirect.net/eric/ConfigKit/internal/document"
	yamladapter "gitea.maximumdirect.net/eric/ConfigKit/internal/yaml"
)

type importAdmissionOutcome uint8

const (
	importAdmissionReady importAdmissionOutcome = iota
	importAdmissionOptionalMissing
)

type admittedImport struct {
	canonical            string
	identity             resourceIdentity
	nodeKey              importNodeKey
	source               SourceInfo
	payload              *document.Node
	declarations         []importDeclarationSpec
	sequenceDeclarations []sequenceDeclarationSpec
	processing           *sourceVersionProcessingResult
}

type importAdmissionResult struct {
	outcome importAdmissionOutcome
	source  admittedImport
}

func acquireImportedInput(request importAcquisitionRequest, policy *acquisitionPolicy, state *acquisitionState, tier *tierAcquisitionState) (*composedDocument, []SourceInfo, *resolvedFilesystemContext, []Diagnostic) {
	source := request.input.input.source
	context := request.input.input.filesystem
	main := request.main
	resolved, diagnostic := resolveFilesystemContext(source, context, main, tier.hostToken)
	if diagnostic != nil {
		return nil, sourceInfos(main.source), nil, []Diagnostic{*diagnostic}
	}
	if resolved.native == nil {
		if diagnostic := validateResolvedFilesystemDirectories(resolved); diagnostic != nil {
			return nil, sourceInfos(main.source), nil, []Diagnostic{*diagnostic}
		}
	}
	access, err := openResolvedFilesystemAccess(resolved)
	if err != nil {
		diagnostic := globFilesystemDiagnostic(err, "configuration root could not be opened", Location{})
		diagnostic.Primary = nil
		return nil, sourceInfos(main.source), nil, []Diagnostic{diagnostic}
	}
	composition, sources, diagnostics := acquireImportedWithAccess(request, policy, state, tier, access, resolved)
	if len(diagnostics) != 0 {
		return composition, sources, nil, diagnostics
	}
	return composition, sources, &resolved, nil
}

func acquireImportedWithAccess(request importAcquisitionRequest, policy *acquisitionPolicy, loadState *acquisitionState, tier *tierAcquisitionState, access *filesystemAccess, resolved resolvedFilesystemContext) (*composedDocument, []SourceInfo, []Diagnostic) {
	main := request.main
	destination := request.input.input.destination
	limits := policy.importLimits
	accounting := loadState.accounting
	versionState := loadState.versionState
	processing := request.processing
	basePath, mainPath := resolved.basePath, resolved.mainPath
	if tier.visited == nil {
		tier.visited = make(map[resourceIdentity]importRoute)
	}
	visited := tier.visited
	if !access.contains(basePath) || (mainPath != "" && !access.contains(mainPath)) {
		_ = access.close()
		return nil, sourceInfos(main.source), []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "selected source and import base must stay inside the configuration root", nil)}
	}
	for _, directory := range []string{access.rootPath, basePath} {
		info, err := access.stat(directory)
		if err != nil || !info.IsDir() {
			_ = access.close()
			return nil, sourceInfos(main.source), []Diagnostic{importDiagnostic(CodeImportInvalid, "configuration root and import base must be directories", nil)}
		}
	}
	mainInfo := cloneSourceInfo(*main.source)
	mainInfo.Destination = destination.clone()
	mainIsItem := main.source.placementMode == placementModeSequenceItem
	if mainPath != "" {
		mainInfo.Path = mainPath
	}
	mainResource := access.identity(mainPath)
	mainNodeKey := importKeyFor(access, mainPath, mainPath == "")
	if mainPath == "" {
		mainResource = access.syntheticIdentity()
	}
	if previous, repeated := visited[mainResource]; repeated {
		_ = access.close()
		return nil, sourceInfos(main.source), []Diagnostic{importRepeatedDiagnostic(Location{SourceID: main.source.ID}, previous, importRoute{})}
	}
	accounting.sources++
	accounting.bytes += main.bytes
	accounting.nodes += countInputNodes(main.document)
	directiveContext := request.input.directiveContext
	if directiveContext == 0 {
		directiveContext = rootDirectiveContextImportsOnly
	}
	if processing != nil && processing.directiveContext != 0 {
		directiveContext = processing.directiveContext
	}
	request.input.directiveContext = directiveContext
	state := importTraversal{
		access: access, request: request, policy: policy, state: loadState, tier: tier,
		scope:       acquisitionScope{owner: mainNodeKey, destination: destination.clone()},
		sources:     []SourceInfo{mainInfo},
		active:      make(map[resourceIdentity]int),
		importGraph: newImportInclusionGraph(),
	}
	if mainIsItem {
		state.itemScope = true
		state.itemPrefix = destination.clone()
		state.itemFallback = destination.clone()
		state.scope.destination = Path{}
		state.sources[0] = cloneSourceInfo(mainInfo)
	}

	if diagnostic := state.checkInitialLimits(); diagnostic != nil {
		_ = access.close()
		return nil, cloneSourceInfos(state.sources), []Diagnostic{*diagnostic}
	}

	mainIdentity := mainPath
	if mainIdentity == "" {
		mainIdentity = "<selected-source>"
	}
	mainRoute := importRoute{canonical: mainIdentity}
	state.importGraph.addNode(importGraphNode{key: mainNodeKey, canonical: mainIdentity, sourceID: mainInfo.ID, role: importRoleMain})
	mainRoute.key = mainNodeKey
	state.active[mainResource] = 0
	tier.visited[mainResource] = mainRoute
	state.stack = append(state.stack, mainRoute)
	var body *document.Node
	var declarations []importDeclarationSpec
	var sequenceDeclarations []sequenceDeclarationSpec
	var diagnostics []Diagnostic
	if processing != nil {
		body = processing.payload
		declarations = processing.imports
		sequenceDeclarations = cloneSequenceDeclarations(processing.sequences)
	} else {
		body, declarations, diagnostics = splitImportDirectives(main.document)
	}
	if len(diagnostics) == 0 {
		if mainIsItem {
			state.scope.contributions = append(state.scope.contributions, acquiredContribution{
				source: cloneSourceInfo(mainInfo), identity: mainNodeKey, role: importRoleMain,
				scopeOwner: mainNodeKey, payload: body, document: body, processing: processing,
				sequenceDeclarations: cloneSequenceDeclarations(sequenceDeclarations),
			})
			diagnostics = state.walkDeclarations(basePath, declarations, 0, Path{})
		} else {
			_, placed, err := placeDocument(body, destination, body.Location, maxDocumentDepth, maxDocumentNodes)
			placementNodes := destinationWrapperNodeCount(destination)
			if err != nil {
				diagnostics = []Diagnostic{sourceDiagnostic(CodeSourceLimitExceeded, main.source.ID, "configuration exceeds the source depth or node limit after destination placement")}
			} else if accounting.nodes+placementNodes > limits.nodes {
				diagnostics = []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", nil)}
			} else {
				accounting.nodes += placementNodes
				state.scope.contributions = append(state.scope.contributions, acquiredContribution{
					source: cloneSourceInfo(mainInfo), payload: body, document: placed,
					identity: mainNodeKey, role: importRoleMain, scopeOwner: mainNodeKey,
					destination: destination.clone(), processing: processing,
					sequenceDeclarations: cloneSequenceDeclarations(sequenceDeclarations),
				})
				diagnostics = state.walkDeclarations(basePath, declarations, 0, destination)
			}
		}
	}
	if len(diagnostics) == 0 {
		state.scheduleAcquisitionScopes(&state.scope)
	}

	closeErr := access.close()
	if len(diagnostics) != 0 {
		if mainIsItem {
			diagnostics = prependCompositionDiagnosticPaths(diagnostics, destination.Index(0))
		}
		return nil, cloneSourceInfos(state.sources), diagnostics
	}
	if closeErr != nil {
		return nil, cloneSourceInfos(state.sources), []Diagnostic{importDiagnostic(CodeImportRead, "configuration root could not be closed", nil)}
	}
	acquisition := state.scope
	if mainIsItem {
		localScope := state.scope
		acquisition = acquisitionScope{
			owner: mainNodeKey, destination: destination.clone(),
			contributions: []acquiredContribution{{
				source: cloneSourceInfo(mainInfo), payload: body, identity: mainNodeKey,
				role: importRoleMain, scopeOwner: mainNodeKey, destination: destination.clone(),
				processing: processing, sequenceDeclarations: cloneSequenceDeclarations(sequenceDeclarations), pendingItem: &localScope,
			}},
		}
	}
	if request.input.deferAssembly {
		if diagnostic := reservePendingItemPlacements(&acquisition, limits, accounting); diagnostic != nil {
			return nil, cloneSourceInfos(state.sources), []Diagnostic{*diagnostic}
		}
		return &composedDocument{
			acquisition: []acquisitionScope{cloneAcquisitionScope(acquisition)},
			importGraph: state.importGraph.clone(), inputBytes: accounting.bytes, inputNodes: accounting.nodes,
		}, cloneSourceInfos(state.sources), nil
	}
	acquired := &composedDocument{
		acquisition: []acquisitionScope{cloneAcquisitionScope(acquisition)},
		importGraph: state.importGraph.clone(), inputBytes: accounting.bytes, inputNodes: accounting.nodes,
	}
	input := assembledInput{
		parsed: main, composition: acquired, acquisition: cloneAcquisitionScopes(acquired.acquisition),
		sources: state.sources, versionState: versionState, accounting: accounting,
	}
	composed, diagnostics := composeAcquiredInputWithPolicies(input, collectionPolicyScope{}, policy.sequencePolicy)
	return composed, cloneSourceInfos(state.sources), diagnostics
}

func (s *importTraversal) checkInitialLimits() *Diagnostic {
	if s.state.accounting.sources > s.policy.importLimits.sources {
		value := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the source-count limit", nil)
		return &value
	}
	if s.state.accounting.bytes > s.policy.importLimits.bytes {
		value := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate byte limit", nil)
		return &value
	}
	if s.state.accounting.nodes > s.policy.importLimits.nodes {
		value := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", nil)
		return &value
	}
	return nil
}

func (s *importTraversal) walkDeclarations(basePath string, declarations []importDeclarationSpec, depth int, parentDestination Path) []Diagnostic {
	policies := make([]resolvedImportPolicy, len(declarations))
	for index, spec := range declarations {
		policy, diagnostics := resolveImportPolicy(spec, s.policy.importDefaults)
		if len(diagnostics) != 0 {
			return diagnostics
		}
		if policy.placement == placementModeSequenceItem && len(appendDestination(parentDestination, policy.destination).segments) == 0 {
			site := spec.fieldSites["placement"]
			return []Diagnostic{{Code: CodeImportInvalid, Severity: SeverityError, Path: site.path.clone(), Message: "sequence-item placement requires a nonempty destination", Primary: cloneLocation(&site.location)}}
		}
		policies[index] = policy
	}
	for index, spec := range declarations {
		policy := policies[index]
		var selection importSelection
		var diagnostics []Diagnostic
		switch spec.selector {
		case importSelectorPath:
			selection, diagnostics = selectNamedImportCandidates(s.access, basePath, spec, policy)
		case importSelectorDirectory:
			selection, diagnostics = selectDirectoryImportCandidates(s.access, basePath, spec, policy, s.policy.importLimits, s.state.accounting)
		case importSelectorGlob:
			selection, diagnostics = selectGlobImportCandidates(s.access, basePath, spec, policy, s.policy.importLimits, s.state.accounting)
		default:
			location := spec.location
			return []Diagnostic{importDiagnostic(CodeImportInvalid, "import selector is invalid", &location)}
		}
		if len(diagnostics) != 0 {
			return diagnostics
		}
		childDestination := appendDestination(parentDestination, policy.destination)
		// Count admitted records rather than lexical candidates: optional missing
		// paths are omitted during admission, and a partially present list is not empty.
		sourcesBefore := len(s.sources)
		for _, candidate := range selection.candidates {
			if diagnostics := s.visitPath(candidate, spec, policy, childDestination, policy.destination, depth+1); len(diagnostics) != 0 {
				return diagnostics
			}
		}
		if !policy.required && len(s.sources) == sourcesBefore {
			s.recordEmptyImportSelection(spec)
		}
	}
	return nil
}

func (s *importTraversal) recordEmptyImportSelection(spec importDeclarationSpec) {
	for index := range s.sources {
		if s.sources[index].ID != spec.location.SourceID {
			continue
		}
		s.sources[index].importSelections = append(s.sources[index].importSelections, importSelectionEvidence{
			path: spec.path.clone(), location: spec.location,
			selector: spec.selector, outcome: importSelectionNone,
		})
		return
	}
}

func (s *importTraversal) admitImportSource(canonical string, identity resourceIdentity, nodeKey importNodeKey, candidate selectedImportCandidate, policy resolvedImportPolicy, destination Path, location Location) (importAdmissionResult, []Diagnostic) {
	sourceID := "source-" + strconv.Itoa(s.state.accounting.sources+1)
	kind := SourceKindFile
	label := candidate.label
	if label == "" {
		label = candidate.path
	}
	if s.access.native != nil {
		kind = SourceKindFilesystem
		label = s.access.label + "/" + canonical
	}
	info := SourceInfo{ID: sourceID, Label: label, Path: canonical, Destination: s.sourceDestination(destination), Layer: LayerBase, Kind: kind, ImportedFrom: &location}
	if policy.placement == placementModeSequenceItem {
		info.placementMode = placementModeSequenceItem
	}
	// A source's own item placement still has an exact sequence-container
	// destination. Only an unresolved enclosing item index makes the path a
	// fallback container for this snapshot.
	info.destinationUnresolved = s.itemScope
	if kind == SourceKindFilesystem {
		info.namespace = s.access.token
	}
	s.sources = append(s.sources, info)
	data, skipped, diagnostic := s.readImport(canonical, sourceID, policy.required)
	if diagnostic != nil {
		return importAdmissionResult{}, []Diagnostic{*diagnostic}
	}
	if skipped {
		s.sources = s.sources[:len(s.sources)-1]
		return importAdmissionResult{outcome: importAdmissionOptionalMissing}, nil
	}
	s.state.accounting.sources++

	root, issues := yamladapter.Parse(bytes.NewReader(data), sourceID, yamladapter.Limits{
		Depth: maxDocumentDepth, Nodes: maxDocumentNodes,
	})
	if len(issues) != 0 {
		return importAdmissionResult{}, parserDiagnostics(sourceID, issues)
	}
	nodeCount := countInputNodes(root)
	if s.state.accounting.nodes+nodeCount > s.policy.importLimits.nodes {
		value := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", &location)
		return importAdmissionResult{}, []Diagnostic{value}
	}
	s.state.accounting.nodes += nodeCount

	directiveContext := s.request.input.directiveContext
	if directiveContext == 0 {
		directiveContext = rootDirectiveContextImportsOnly
	}
	versionDestination := destination
	if s.itemScope {
		versionDestination = appendDestination(s.itemPrefix.Index(0), destination)
	}
	var processing sourceVersionProcessingResult
	if s.state.versionState != nil && s.state.versionState.mode != sourceVersionModeDefault {
		processed, failures := s.state.versionState.processWithDirectiveContext(root, sourceVersionContribution, versionDestination, sourceID, policy.placement, directiveContext)
		if len(failures) != 0 {
			return importAdmissionResult{}, failures
		}
		processing = processed
	} else {
		processed := processSourceVersionWithDirectiveContext(root, s.request.input.protectionRule, sourceVersionContribution, versionDestination, sourceVersionModeDefault, directiveContext)
		if len(processed.diagnostics) != 0 {
			return importAdmissionResult{}, processed.diagnostics
		}
		processing = processed
	}
	return importAdmissionResult{source: admittedImport{
		canonical: canonical, identity: identity, nodeKey: nodeKey,
		source: cloneSourceInfo(info), payload: processing.payload,
		declarations:         processing.imports,
		sequenceDeclarations: cloneSequenceDeclarations(processing.sequences),
		processing:           &processing,
	}}, nil
}

func (s *importTraversal) visitPath(candidate selectedImportCandidate, spec importDeclarationSpec, policy resolvedImportPolicy, destination, relativeDestination Path, depth int) []Diagnostic {
	location := candidate.site.location
	if location.SourceID == "" && location.Line == 0 && location.Column == 0 {
		location = spec.location
	}
	if !s.access.contains(candidate.accessPath) {
		return []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "import target is outside the configuration root", &location)}
	}
	canonical, err := s.access.canonical(candidate.accessPath)
	if err != nil {
		if !policy.required && errors.Is(err, fs.ErrNotExist) {
			return nil
		}
		code := CodeImportInvalid
		message := "import target could not be canonicalized"
		if errors.Is(err, fs.ErrNotExist) {
			code, message = CodeImportMissing, "required import target does not exist"
		} else if errors.Is(err, fs.ErrPermission) {
			message = "import target cannot be accessed"
		}
		return []Diagnostic{importDiagnostic(code, message, &location)}
	}
	if !s.access.contains(canonical) {
		return []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "import target resolves outside the configuration root", &location)}
	}
	if depth > s.policy.importLimits.depth {
		return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the active import-depth limit", &location)}
	}

	currentRoute := s.currentRoute(location)
	identity := s.access.identity(canonical)
	nodeKey := importKeyFor(s.access, canonical, false)
	importerKey := s.currentImportNode()
	if activeIndex, ok := s.active[identity]; ok {
		closingEdge := s.importGraph.addEdge(importerKey, nodeKey, location)
		if diagnostic, ok := s.importGraph.scheduleCycle(closingEdge, location); ok {
			return []Diagnostic{diagnostic}
		}
		cycle := append([]importRoute(nil), s.stack[activeIndex:]...)
		return []Diagnostic{importCycleDiagnostic(canonical, location, cycle)}
	}
	if previous, ok := s.tier.visited[identity]; ok {
		s.importGraph.addEdge(importerKey, nodeKey, location)
		return []Diagnostic{importRepeatedDiagnostic(location, previous, currentRoute)}
	}
	if s.state.accounting.sources+1 > s.policy.importLimits.sources {
		return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the source-count limit", &location)}
	}

	if _, err := s.access.relative(canonical); err != nil {
		return []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "import target could not be addressed inside the configuration root", &location)}
	}
	admission, diagnostics := s.admitImportSource(canonical, identity, nodeKey, candidate, policy, destination, location)
	if len(diagnostics) != 0 {
		return diagnostics
	}
	if admission.outcome == importAdmissionOptionalMissing {
		return nil
	}
	admitted := admission.source

	placementDestination := destination
	mode := placementModeMapping
	if policy.placement == placementModeSequenceItem {
		mode = placementModeSequenceItem
	}
	role := importRoleMapping
	if mode == placementModeSequenceItem {
		role = importRoleItem
	}
	s.importGraph.addNode(importGraphNode{key: admitted.nodeKey, canonical: admitted.canonical, sourceID: admitted.source.ID, role: role})
	s.importGraph.addEdge(importerKey, admitted.nodeKey, location)
	if mode == placementModeSequenceItem {
		route := importRoute{canonical: admitted.canonical, locations: currentRoute.locations, key: admitted.nodeKey}
		s.active[admitted.identity] = len(s.stack)
		s.tier.visited[admitted.identity] = route
		s.stack = append(s.stack, route)
		diagnostics := s.visitItemSource(admitted, destination, relativeDestination, depth)
		delete(s.active, admitted.identity)
		s.stack = s.stack[:len(s.stack)-1]
		return diagnostics
	}
	_, placed, err := placeDocument(admitted.payload, placementDestination, admitted.payload.Location, maxDocumentDepth, maxDocumentNodes)
	if err != nil {
		return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the source depth or node limit after destination placement", &location)}
	}
	placementNodes := destinationWrapperNodeCount(placementDestination)
	if s.state.accounting.nodes+placementNodes > s.policy.importLimits.nodes {
		return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", &location)}
	}
	s.state.accounting.nodes += placementNodes
	route := importRoute{canonical: admitted.canonical, locations: currentRoute.locations, key: admitted.nodeKey}
	s.scope.contributions = append(s.scope.contributions, acquiredContribution{
		source: cloneSourceInfo(admitted.source), payload: admitted.payload, document: placed,
		identity: admitted.nodeKey, role: role, scopeOwner: s.scope.owner,
		destination: placementDestination.clone(), relativeDestination: relativeDestination.clone(), processing: admitted.processing,
		sequenceDeclarations: cloneSequenceDeclarations(admitted.sequenceDeclarations),
	})
	s.active[admitted.identity] = len(s.stack)
	s.tier.visited[admitted.identity] = route
	s.stack = append(s.stack, route)
	diagnostics = s.walkDeclarations(s.access.dir(admitted.canonical), admitted.declarations, depth, destination)
	delete(s.active, admitted.identity)
	s.stack = s.stack[:len(s.stack)-1]
	return diagnostics
}

func (s *importTraversal) currentImportNode() importNodeKey {
	if len(s.stack) != 0 {
		return s.stack[len(s.stack)-1].key
	}
	key := s.scope.owner
	if _, exists := s.importGraph.index[key]; !exists {
		s.importGraph.addNode(importGraphNode{key: key, canonical: "<selected-source>", role: importRoleMain})
	}
	return key
}

func (s *importTraversal) sourceDestination(destination Path) Path {
	if !s.itemScope {
		return destination.clone()
	}
	// Until the enclosing item receives its assigned index, only the collection
	// container is exact. Keep the local suffix in the acquisition projection.
	return s.itemFallback.clone()
}

func (s *importTraversal) readImport(name, sourceID string, required bool) ([]byte, bool, *Diagnostic) {
	remaining := s.policy.importLimits.bytes - s.state.accounting.bytes
	limit := min(int64(maxSourceBytes), remaining)
	if limit < 0 {
		limit = 0
	}
	data, failure := readRegularFile(name, s.access.stat, s.access.open, limit)
	if failure != nil {
		if !required && failure.missingTarget {
			return nil, true, nil
		}
		value := sourceReadDiagnostic(failure, sourceID, CodeImportLimitExceeded, "configuration exceeds a source or aggregate byte limit")
		return nil, false, &value
	}
	s.state.accounting.bytes += int64(len(data))
	return data, false, nil
}

func (s *importTraversal) currentRoute(location Location) importRoute {
	locations := make([]Location, 0, len(s.stack)+1)
	for _, route := range s.stack {
		if len(route.locations) != 0 {
			locations = append(locations, route.locations[len(route.locations)-1])
		}
	}
	locations = append(locations, location)
	return importRoute{locations: locations}
}