package plan

import (
	"fmt"
	"sort"
)

// fanOutScopeAnalysis is the compiler-proven lexical body of one dynamic
// fan-out. Runtime consumes this closed scope instead of rediscovering body
// membership or join ownership from arbitrary graph reachability.
type fanOutScopeAnalysis struct {
	ownerID         string
	joinID          string
	members         []string
	bodyEntryEdges  []string
	successArrivals []string
	failedTerminals []string
}

type exploredFanOutScope struct {
	nodes           map[string]bool
	boundaries      []*Edge
	failedTerminals map[string]bool
	succeededEnds   map[string]bool
}

func (v *validator) validateFanOutScopes(outgoing, incoming map[string]map[string][]*Edge) {
	owners := make([]*Node, 0)
	joinsByOwner := map[string][]*Node{}
	for _, node := range v.nodes {
		if node.Type == NodeFanOut && node.FanOut != nil {
			owners = append(owners, node)
		}
		if node.Type == NodeJoin && node.Join != nil && node.Join.FanOutID != "" {
			joinsByOwner[node.Join.FanOutID] = append(joinsByOwner[node.Join.FanOutID], node)
		}
	}
	sort.Slice(owners, func(i, j int) bool { return owners[i].ID < owners[j].ID })
	for ownerID := range joinsByOwner {
		sort.Slice(joinsByOwner[ownerID], func(i, j int) bool {
			return joinsByOwner[ownerID][i].ID < joinsByOwner[ownerID][j].ID
		})
	}

	reachable := v.reachableNodes()
	dominators := v.dominators(reachable)
	for _, owner := range owners {
		ownerPath := fmt.Sprintf("graph.nodes[%d]", v.nodeIndexes[owner.ID])
		joins := joinsByOwner[owner.ID]
		if !reachable[owner.ID] {
			v.error("fan_out.unreachable_owner", ownerPath, fmt.Sprintf("fan_out %q is not reachable from entry", owner.ID), owner.ID, "")
		}
		if len(joins) == 0 {
			v.error("fan_out.join_required", ownerPath, fmt.Sprintf("fan_out %q requires one closing join", owner.ID), owner.ID, "")
			continue
		}
		if len(joins) > 1 {
			v.error("fan_out.multiple_joins", ownerPath, fmt.Sprintf("fan_out %q has %d joins; exactly one is allowed", owner.ID, len(joins)), owner.ID, "")
			continue
		}
		join := joins[0]
		if !dominators[join.ID][owner.ID] {
			v.error("fan_out.undominated_join", fmt.Sprintf("graph.nodes[%d].join.fanOutId", v.nodeIndexes[join.ID]), fmt.Sprintf("fan_out %q does not dominate join %q", owner.ID, join.ID), join.ID, "")
		}

		scope := v.exploreFanOutBody(owner, join, outgoing)
		for nodeID := range scope.nodes {
			if nested := v.nodes[nodeID]; nested != nil && nested.Type == NodeFanOut {
				v.error(
					"fan_out.nested_not_allowed",
					fmt.Sprintf("graph.nodes[%d]", v.nodeIndexes[nodeID]),
					fmt.Sprintf("fan_out %q cannot be nested inside fan_out %q", nodeID, owner.ID),
					nodeID,
					"",
				)
			}
			if reachable[nodeID] && !dominators[nodeID][owner.ID] {
				v.error("fan_out.undominated_body", fmt.Sprintf("graph.nodes[%d]", v.nodeIndexes[nodeID]), fmt.Sprintf("fan_out %q does not dominate body node %q", owner.ID, nodeID), nodeID, "")
			}
		}
		if scope.nodes[v.document.Graph.Entry.NodeID] {
			v.error("fan_out.entry_into_body", "graph.entry.nodeId", fmt.Sprintf("graph entry cannot target body node %q of fan_out %q", v.document.Graph.Entry.NodeID, owner.ID), v.document.Graph.Entry.NodeID, "")
		}
		if len(scope.boundaries) == 0 || len(scope.succeededEnds) != 0 {
			v.error("fan_out.open_body", ownerPath, fmt.Sprintf("fan_out %q has a path that does not close at join %q or an item failure", owner.ID, join.ID), owner.ID, "")
		}
		v.validateFanOutMembership(owner, join, scope, incoming)
		v.validateFanOutStateEffects(owner, scope)
		v.recordFanOutScope(owner, join, scope, outgoing)
	}
}

func (v *validator) exploreFanOutBody(owner, join *Node, outgoing map[string]map[string][]*Edge) *exploredFanOutScope {
	scope := &exploredFanOutScope{
		nodes:           map[string]bool{},
		failedTerminals: map[string]bool{},
		succeededEnds:   map[string]bool{},
	}
	queue := make([]string, 0)
	process := func(edge *Edge) {
		if edge == nil {
			return
		}
		if edge.Target.NodeID == join.ID {
			scope.boundaries = append(scope.boundaries, edge)
			if edge.Target.Port != "item" {
				v.error("fan_out.cross_wired_item", fmt.Sprintf("graph.edges[%d].target.port", v.edgeIndexes[edge.ID]), fmt.Sprintf("fan_out item reaches join port %q", edge.Target.Port), owner.ID, edge.ID)
			}
			return
		}
		if _, exists := v.nodes[edge.Target.NodeID]; exists {
			queue = append(queue, edge.Target.NodeID)
		}
	}
	for _, edge := range outgoing[owner.ID]["item"] {
		process(edge)
	}
	for len(queue) != 0 {
		current := queue[0]
		queue = queue[1:]
		if scope.nodes[current] {
			continue
		}
		scope.nodes[current] = true
		node := v.nodes[current]
		if node == nil {
			continue
		}
		if node.Type == NodeEnd {
			if node.End != nil && node.End.Outcome == EndFailed {
				scope.failedTerminals[node.ID] = true
			} else {
				scope.succeededEnds[node.ID] = true
			}
			continue
		}
		ports := sourcePorts(node)
		sort.Strings(ports)
		for _, port := range ports {
			edges := append([]*Edge(nil), outgoing[current][port]...)
			sort.Slice(edges, func(i, j int) bool { return edges[i].ID < edges[j].ID })
			// An omitted recoverable failure port lowers to a fan-out boundary
			// transition. It is a closed item-failure path, not an open body.
			if len(edges) == 0 && isImplicitFailurePort(port) {
				continue
			}
			for _, edge := range edges {
				process(edge)
			}
		}
	}
	sort.Slice(scope.boundaries, func(i, j int) bool { return scope.boundaries[i].ID < scope.boundaries[j].ID })
	return scope
}

func (v *validator) validateFanOutMembership(owner, join *Node, scope *exploredFanOutScope, incoming map[string]map[string][]*Edge) {
	for nodeID := range scope.nodes {
		for _, edges := range incoming[nodeID] {
			for _, edge := range edges {
				fromOwner := edge.Source.NodeID == owner.ID && edge.Source.Port == "item"
				if !fromOwner && !scope.nodes[edge.Source.NodeID] {
					v.error("fan_out.external_entry", fmt.Sprintf("graph.edges[%d]", v.edgeIndexes[edge.ID]), fmt.Sprintf("fan_out body node %q has an incoming edge from outside fan_out %q", nodeID, owner.ID), nodeID, edge.ID)
				}
			}
		}
	}
	for _, edge := range incoming[join.ID]["item"] {
		fromOwner := edge.Source.NodeID == owner.ID && edge.Source.Port == "item"
		if !fromOwner && !scope.nodes[edge.Source.NodeID] {
			v.error("fan_out.external_arrival", fmt.Sprintf("graph.edges[%d]", v.edgeIndexes[edge.ID]), fmt.Sprintf("fan_out join %q receives an item from outside fan_out %q", join.ID, owner.ID), join.ID, edge.ID)
		}
	}
}

func (v *validator) validateFanOutStateEffects(owner *Node, scope *exploredFanOutScope) {
	effects := compileDocumentStateEffects(v)
	for nodeID := range scope.nodes {
		if len(effects[nodeID].Writes) == 0 {
			continue
		}
		v.error("fan_out.state_write_not_allowed", fmt.Sprintf("graph.nodes[%d]", v.nodeIndexes[nodeID]), fmt.Sprintf("fan_out body node %q cannot write Run state because items execute concurrently", nodeID), nodeID, "")
	}
}

func (v *validator) recordFanOutScope(owner, join *Node, scope *exploredFanOutScope, outgoing map[string]map[string][]*Edge) {
	snapshot := fanOutScopeAnalysis{
		ownerID:         owner.ID,
		joinID:          join.ID,
		members:         []string{},
		bodyEntryEdges:  []string{},
		successArrivals: []string{},
		failedTerminals: []string{},
	}
	for nodeID := range scope.nodes {
		node := v.nodes[nodeID]
		if node != nil && node.Type != NodeEnd {
			snapshot.members = append(snapshot.members, nodeID)
		}
	}
	for _, edge := range outgoing[owner.ID]["item"] {
		snapshot.bodyEntryEdges = append(snapshot.bodyEntryEdges, edge.ID)
	}
	for _, edge := range scope.boundaries {
		snapshot.successArrivals = append(snapshot.successArrivals, edge.ID)
	}
	for nodeID := range scope.failedTerminals {
		snapshot.failedTerminals = append(snapshot.failedTerminals, nodeID)
	}
	sort.Strings(snapshot.members)
	sort.Strings(snapshot.bodyEntryEdges)
	sort.Strings(snapshot.successArrivals)
	sort.Strings(snapshot.failedTerminals)
	v.fanOutScopes = append(v.fanOutScopes, snapshot)
}