package plan

import (
	"fmt"
	"sort"
	"strings"
)

// work_items is a nominal authorization-bearing value, not an ordinary
// business schema. The compiler creates exactly one source for it:
// output.workItems on a callback that declares sendMessage.workItems. Authors
// may declare the direct fan_out activation input that receives that value,
// but no Plan contract, callback business output, state, or intermediate node
// may manufacture or retain another work_items value.
func (v *validator) rejectAuthoredWorkItemsSchemas(path string, schemas map[string]Schema, nodeID string) {
	for _, name := range sortedSchemaKeys(schemas) {
		v.rejectAuthoredWorkItemsSchema(path+"."+name, schemas[name], nodeID)
	}
}

func (v *validator) rejectAuthoredWorkItemsSchema(path string, schema Schema, nodeID string) {
	if schema.Type == TypeWorkItems {
		v.error(
			"schema.work_items_reserved",
			path,
			"work_items is reserved for compiler-owned callback output and the direct fan_out input that consumes it",
			nodeID,
			"",
		)
	}
	if schema.Items != nil {
		v.rejectAuthoredWorkItemsSchema(path+".items", *schema.Items, nodeID)
	}
}

func schemaContainsWorkItems(schema Schema) bool {
	if schema.Type == TypeWorkItems {
		return true
	}
	return schema.Items != nil && schemaContainsWorkItems(*schema.Items)
}

func sortedWorkItemNodeIDs(nodes map[string]*Node) []string {
	ids := make([]string, 0, len(nodes))
	for id := range nodes {
		ids = append(ids, id)
	}
	sort.Strings(ids)
	return ids
}

func (v *validator) validateNodeWorkItemsInput(path string, node *Node) {
	if node == nil {
		return
	}
	if node.Type != NodeFanOut || node.FanOut == nil {
		v.rejectAuthoredWorkItemsSchemas(path+".input", node.Input, node.ID)
		return
	}

	field, direct := directFanOutInputReference(node.FanOut.Items)
	if !direct {
		v.error(
			"fan_out.items_source",
			path+".fanOut.items",
			"fanOut.items must directly reference the fan_out input populated from compiler-owned callback output.workItems",
			node.ID,
			"",
		)
	}
	for _, name := range sortedSchemaKeys(node.Input) {
		schema := node.Input[name]
		if !schemaContainsWorkItems(schema) {
			continue
		}
		if !direct || name != field || schema.Type != TypeWorkItems || schema.Items != nil {
			v.rejectAuthoredWorkItemsSchema(path+".input."+name, schema, node.ID)
		}
	}
}

func directFanOutInputReference(expression ValueExpr) (string, bool) {
	if expression.Value != nil || strings.TrimSpace(expression.Expr) != "" {
		return "", false
	}
	segments := strings.Split(strings.TrimSpace(expression.Ref), ".")
	if len(segments) != 2 || segments[0] != "input" || !idPattern.MatchString(segments[1]) {
		return "", false
	}
	return segments[1], true
}

func directCompilerWorkItemsOutputReference(expression ValueExpr) bool {
	return expression.Value == nil &&
		strings.TrimSpace(expression.Expr) == "" &&
		strings.TrimSpace(expression.Ref) == "output."+sendMessageOutputWorkItems
}

func (v *validator) validateFanOutWorkItemsSources(
	incoming,
	outgoing map[string]map[string][]*Edge,
) {
	for _, nodeID := range sortedWorkItemNodeIDs(v.nodes) {
		node := v.nodes[nodeID]
		if node == nil || node.Type != NodeFanOut || node.FanOut == nil {
			continue
		}
		field, direct := directFanOutInputReference(node.FanOut.Items)
		if !direct {
			continue
		}
		path := fmt.Sprintf("graph.nodes[%d].fanOut.items", v.nodeIndexes[node.ID])
		if v.document.Graph.Entry.NodeID == node.ID && v.document.Graph.Entry.Port == "in" {
			v.error(
				"fan_out.items_source",
				path,
				"fan_out cannot be a graph entry because WorkItems must originate from a controlled callback",
				node.ID,
				"",
			)
		}
		incomingEdges := incoming[node.ID]["in"]
		if len(incomingEdges) == 0 {
			v.error(
				"fan_out.items_source",
				path,
				"fan_out requires an incoming controlled callback WorkItems source",
				node.ID,
				"",
			)
		}
		for _, edge := range incomingEdges {
			source := v.nodes[edge.Source.NodeID]
			bindings := edge.Bindings
			if bindings == nil && source != nil {
				sourceSchema, root, exposesPayload := sourcePayloadContract(source, edge.Source.Port)
				if exposesPayload {
					bindings, _ = omittedBindings(sourceSchema, root, node.Input)
				}
			}
			expression, bound := bindings[field]
			if !bound {
				edgePath := fmt.Sprintf("graph.edges[%d].bindings.%s", v.edgeIndexes[edge.ID], field)
				v.error(
					"fan_out.items_source",
					edgePath,
					"fan_out WorkItems must be bound directly from output.workItems of a callback declaring sendMessage.workItems",
					node.ID,
					edge.ID,
				)
				continue
			}
			controlledSource := source != nil &&
				source.Type == NodeSendMessage &&
				source.SendMessage != nil &&
				source.SendMessage.Completion == CompletionCallback &&
				source.SendMessage.WorkItems != nil &&
				edge.Source.Port == "completed"
			if controlledSource && directCompilerWorkItemsOutputReference(expression) {
				continue
			}
			edgePath := fmt.Sprintf("graph.edges[%d].bindings.%s", v.edgeIndexes[edge.ID], field)
			v.error(
				"fan_out.items_source",
				edgePath,
				"fan_out WorkItems must be bound directly from output.workItems of a callback declaring sendMessage.workItems",
				node.ID,
				edge.ID,
			)
		}
	}

	// A Planner can author dependency references only when one callback has one
	// statically knowable fan_out item-result contract. Keep the authoring
	// boundary strict instead of letting get_current guess which downstream
	// branch defines dependency.value.
	for _, nodeID := range sortedWorkItemNodeIDs(v.nodes) {
		node := v.nodes[nodeID]
		if node == nil || node.Type != NodeSendMessage || node.SendMessage == nil ||
			node.SendMessage.Completion != CompletionCallback || node.SendMessage.WorkItems == nil {
			continue
		}
		consumers := 0
		for _, edge := range outgoing[node.ID]["completed"] {
			target := v.nodes[edge.Target.NodeID]
			if target == nil || target.Type != NodeFanOut || target.FanOut == nil || edge.Target.Port != "in" {
				continue
			}
			field, direct := directFanOutInputReference(target.FanOut.Items)
			if !direct {
				continue
			}
			bindings := edge.Bindings
			if bindings == nil {
				sourceSchema, root, exposesPayload := sourcePayloadContract(node, edge.Source.Port)
				if exposesPayload {
					bindings, _ = omittedBindings(sourceSchema, root, target.Input)
				}
			}
			if expression, ok := bindings[field]; ok && directCompilerWorkItemsOutputReference(expression) {
				consumers++
			}
		}
		if consumers != 1 {
			v.error(
				"send_message.work_items_consumer",
				fmt.Sprintf("graph.nodes[%d].sendMessage.workItems", v.nodeIndexes[node.ID]),
				fmt.Sprintf("WorkItems callback %q must feed exactly one direct fan_out so dependency value contract is unambiguous; found %d", node.ID, consumers),
				node.ID,
				"",
			)
		}
	}
}

func artifactSchemaContainsWorkItems(schema ArtifactSchema) bool {
	if schema.Type == TypeWorkItems {
		return true
	}
	return schema.Items != nil && artifactSchemaContainsWorkItems(*schema.Items)
}

func artifactContractContainsWorkItems(contract ArtifactContract) bool {
	for _, field := range contract {
		if artifactSchemaContainsWorkItems(field.Schema) {
			return true
		}
	}
	return false
}

func directArtifactFanOutInput(expression ArtifactExpression) (string, bool) {
	if expression.Op != ArtifactOpLoad ||
		expression.ResultType != TypeWorkItems ||
		expression.Load == nil ||
		expression.Load.Space != ArtifactLoadActivationInput ||
		len(expression.Load.Path) != 1 {
		return "", false
	}
	return expression.Load.Path[0], true
}

func directArtifactCallbackWorkItemsOutput(expression ArtifactExpression) bool {
	return expression.Op == ArtifactOpLoad &&
		expression.ResultType == TypeWorkItems &&
		expression.Load != nil &&
		expression.Load.Space == ArtifactLoadActivationOutput &&
		len(expression.Load.Path) == 1 &&
		expression.Load.Path[0] == sendMessageOutputWorkItems
}

func validateArtifactWorkItemsContracts(contracts ArtifactContracts) error {
	if artifactContractContainsWorkItems(contracts.Input) {
		return fmt.Errorf("contracts.input cannot declare work_items")
	}
	if artifactContractContainsWorkItems(contracts.Output) {
		return fmt.Errorf("contracts.output cannot declare work_items")
	}
	if artifactContractContainsWorkItems(contracts.State) {
		return fmt.Errorf("contracts.state cannot declare work_items")
	}
	return nil
}

// validateArtifactWorkItemsProvenance defends the immutable artifact boundary
// independently of source validation. A forged or future producer cannot mark
// arbitrary business data as work_items and thereby skip callback responder
// and scope authorization.
func validateArtifactWorkItemsProvenance(artifact *Artifact, nodes map[string]*ArtifactNode) error {
	if err := validateArtifactWorkItemsContracts(artifact.Contracts); err != nil {
		return err
	}

	fanOutFields := make(map[string]string)
	for index := range artifact.Nodes {
		node := &artifact.Nodes[index]
		fanOutField := ""
		if node.Opcode == OpcodeFanOut && node.Config.FanOut != nil {
			var ok bool
			fanOutField, ok = directArtifactFanOutInput(node.Config.FanOut.Items)
			if !ok {
				return fmt.Errorf("node %q fan_out items must directly load its controlled activation input", node.ID)
			}
			fanOutFields[node.ID] = fanOutField
		}

		for _, port := range node.InputPorts {
			for _, field := range port.Contract {
				if !artifactSchemaContainsWorkItems(field.Schema) {
					continue
				}
				if node.Opcode != OpcodeFanOut ||
					port.Name != "in" ||
					field.Name != fanOutField ||
					field.Schema.Type != TypeWorkItems ||
					field.Schema.Items != nil {
					return fmt.Errorf("node %q input %q.%s cannot declare work_items", node.ID, port.Name, field.Name)
				}
			}
		}
		for _, port := range node.OutputPorts {
			for _, field := range port.Contract {
				if !artifactSchemaContainsWorkItems(field.Schema) {
					continue
				}
				controlledOutput := node.Opcode == OpcodeSendMessageCallback &&
					node.Config.SendMessage != nil &&
					node.Config.SendMessage.WorkItems != nil &&
					port.Name == "completed" &&
					field.Name == sendMessageOutputWorkItems &&
					field.Schema.Type == TypeWorkItems &&
					field.Schema.Items == nil
				if !controlledOutput {
					return fmt.Errorf("node %q output %q.%s cannot declare work_items", node.ID, port.Name, field.Name)
				}
			}
		}
		if config := node.Config.SendMessage; config != nil &&
			artifactContractContainsWorkItems(config.CallbackOutput) {
			return fmt.Errorf("node %q sendMessage.callbackOutput cannot declare work_items", node.ID)
		}
		if config := node.Config.Join; config != nil && artifactContractContainsWorkItems(config.Result) {
			return fmt.Errorf("node %q join.result cannot declare work_items", node.ID)
		}
		if config := node.Config.Loop; config != nil && artifactContractContainsWorkItems(config.State) {
			return fmt.Errorf("node %q loop.state cannot declare work_items", node.ID)
		}
	}

	if field, fanOutEntry := fanOutFields[artifact.Entry.NodeID]; fanOutEntry {
		return fmt.Errorf("fan_out node %q input %q cannot be populated by graph entry", artifact.Entry.NodeID, field)
	}
	fanOutNodeIDs := make([]string, 0, len(fanOutFields))
	for nodeID := range fanOutFields {
		fanOutNodeIDs = append(fanOutNodeIDs, nodeID)
	}
	sort.Strings(fanOutNodeIDs)
	for _, nodeID := range fanOutNodeIDs {
		field := fanOutFields[nodeID]
		foundIncoming := false
		for index := range artifact.Edges {
			edge := &artifact.Edges[index]
			if edge.Target.NodeID != nodeID || edge.Target.Port != "in" {
				continue
			}
			foundIncoming = true
			var expression *ArtifactExpression
			for bindingIndex := range edge.Bindings {
				if edge.Bindings[bindingIndex].Target == field {
					expression = &edge.Bindings[bindingIndex].Value
					break
				}
			}
			if expression == nil {
				continue // Required-binding validation reports this separately.
			}
			source := nodes[edge.Source.NodeID]
			controlledSource := source != nil &&
				source.Opcode == OpcodeSendMessageCallback &&
				source.Config.SendMessage != nil &&
				source.Config.SendMessage.WorkItems != nil &&
				edge.Source.Port == "completed"
			if !controlledSource || !directArtifactCallbackWorkItemsOutput(*expression) {
				return fmt.Errorf(
					"edge %q fan_out WorkItems must load compiler-owned callback output.workItems",
					edge.ID,
				)
			}
		}
		if !foundIncoming {
			return fmt.Errorf("fan_out node %q has no controlled callback WorkItems source", nodeID)
		}
	}
	for index := range artifact.Nodes {
		node := &artifact.Nodes[index]
		if node.Opcode != OpcodeSendMessageCallback || node.Config.SendMessage == nil || node.Config.SendMessage.WorkItems == nil {
			continue
		}
		if _, err := WorkItemsDependencyValueContract(artifact, node.ID); err != nil {
			return err
		}
	}
	return nil
}

// WorkItemsDependencyValueContract returns the exact compiled contract stored
// under every terminal dependency envelope's value field for WorkItems
// produced by callbackNodeID. Callers must pass an artifact that has already
// crossed the ordinary envelope validation boundary; this helper nevertheless
// fails closed when the callback does not have one direct fan_out consumer.
//
// The contract is the closing fan_out Join's item input contract, because the
// edge arriving there is the compiler-typed value frozen into each dependency
// envelope. The returned slice and nested schema values are detached from the
// immutable artifact so reducers can freeze them into an Intent safely.
func WorkItemsDependencyValueContract(artifact *Artifact, callbackNodeID string) (ArtifactContract, error) {
	if artifact == nil {
		return nil, fmt.Errorf("WorkItems dependency contract requires an artifact")
	}
	nodes := make(map[string]*ArtifactNode, len(artifact.Nodes))
	for index := range artifact.Nodes {
		node := &artifact.Nodes[index]
		nodes[node.ID] = node
	}
	callback := nodes[callbackNodeID]
	if callback == nil || callback.Opcode != OpcodeSendMessageCallback || callback.Config.SendMessage == nil || callback.Config.SendMessage.WorkItems == nil {
		return nil, fmt.Errorf("node %q is not a WorkItems callback", callbackNodeID)
	}

	consumers := make([]string, 0, 1)
	for index := range artifact.Edges {
		edge := &artifact.Edges[index]
		if edge.Source.NodeID != callbackNodeID || edge.Source.Port != "completed" || edge.Target.Port != "in" {
			continue
		}
		target := nodes[edge.Target.NodeID]
		if target == nil || target.Opcode != OpcodeFanOut || target.Config.FanOut == nil {
			continue
		}
		field, direct := directArtifactFanOutInput(target.Config.FanOut.Items)
		if !direct {
			continue
		}
		for bindingIndex := range edge.Bindings {
			binding := &edge.Bindings[bindingIndex]
			if binding.Target == field && directArtifactCallbackWorkItemsOutput(binding.Value) {
				consumers = append(consumers, target.ID)
				break
			}
		}
	}
	if len(consumers) != 1 {
		return nil, fmt.Errorf(
			"WorkItems callback %q must feed exactly one direct fan_out so dependency value contract is unambiguous; found %d",
			callbackNodeID,
			len(consumers),
		)
	}

	fanOutID := consumers[0]
	joinID := ""
	for index := range artifact.Scopes {
		scope := &artifact.Scopes[index]
		if scope.Kind != ArtifactScopeFanOut || scope.OwnerID != fanOutID || scope.FanOut == nil {
			continue
		}
		if joinID != "" {
			return nil, fmt.Errorf("fan_out %q has multiple compiled scopes", fanOutID)
		}
		joinID = scope.FanOut.JoinID
	}
	join := nodes[joinID]
	if joinID == "" || join == nil || join.Config.Join == nil || join.Config.Join.OwnerID != fanOutID {
		return nil, fmt.Errorf("fan_out %q has no unique closing join", fanOutID)
	}
	contract, ok := artifactPortContractByName(join.InputPorts, "item")
	if !ok {
		return nil, fmt.Errorf("fan_out %q closing join %q has no item input contract", fanOutID, joinID)
	}
	return cloneArtifactContract(contract), nil
}

func artifactPortContractByName(ports []ArtifactPort, name string) (ArtifactContract, bool) {
	for index := range ports {
		if ports[index].Name == name {
			return ports[index].Contract, true
		}
	}
	return nil, false
}

func cloneArtifactContract(contract ArtifactContract) ArtifactContract {
	cloned := make(ArtifactContract, len(contract))
	for index := range contract {
		cloned[index] = ArtifactField{
			Name:   contract[index].Name,
			Schema: cloneArtifactSchema(contract[index].Schema),
		}
	}
	return cloned
}

func cloneArtifactSchema(schema ArtifactSchema) ArtifactSchema {
	cloned := schema
	cloned.Values = append([]string{}, schema.Values...)
	if schema.Default != nil {
		value := append(ArtifactLiteral(nil), (*schema.Default)...)
		cloned.Default = &value
	}
	if schema.Items != nil {
		items := cloneArtifactSchema(*schema.Items)
		cloned.Items = &items
	}
	return cloned
}