package plan

import (
	"fmt"
	"sort"
)

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

// parallelScopeAnalysis is the immutable result of the same traversal that
// proves fork/join correctness. Lowering consumes this snapshot directly so
// the runtime artifact never re-discovers scope semantics from the graph.
type parallelScopeAnalysis struct {
	ownerID    string
	joinID     string
	mode       JoinMode
	activation string
	branches   []parallelBranchAnalysis
}

type parallelBranchAnalysis struct {
	port            string
	members         []string
	entryEdges      []string
	successArrivals []string
	failedTerminals []string
}

func (v *validator) validateParallelScopes(outgoing, incoming map[string]map[string][]*Edge) {
	owners := make([]*Node, 0)
	for _, node := range v.nodes {
		if node.Type == NodeFork && node.Fork != nil || node.Type == NodeSwitch && node.Switch != nil && node.Switch.Mode == SwitchAllMatch {
			owners = append(owners, node)
		}
	}
	if len(owners) == 0 {
		return
	}
	sort.Slice(owners, func(i, j int) bool { return owners[i].ID < owners[j].ID })

	joinsByOwner := map[string][]*Node{}
	for _, node := range v.nodes {
		if node.Type == NodeJoin && node.Join != nil {
			joinsByOwner[node.Join.ForkID] = append(joinsByOwner[node.Join.ForkID], node)
		}
	}
	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("parallel.unreachable_owner", ownerPath, fmt.Sprintf("parallel owner %q is not reachable from entry", owner.ID), owner.ID, "")
		}
		if len(joins) == 0 {
			v.error("parallel.join_required", ownerPath, fmt.Sprintf("parallel owner %q requires one closing join", owner.ID), owner.ID, "")
			continue
		}
		if len(joins) > 1 {
			v.error("parallel.multiple_joins", ownerPath, fmt.Sprintf("parallel owner %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("parallel.undominated_join", fmt.Sprintf("graph.nodes[%d].join.forkId", v.nodeIndexes[join.ID]), fmt.Sprintf("parallel owner %q does not dominate join %q", owner.ID, join.ID), join.ID, "")
		}
		branches, ok := v.joinOwnerBranches(owner.ID)
		if !ok {
			continue
		}
		scopes := map[string]*parallelBranchScope{}
		for _, branch := range branches {
			scope := v.exploreParallelBranch(owner, join, branch, outgoing)
			scopes[branch] = scope
			if len(scope.boundaries) == 0 || len(scope.succeededEnds) != 0 {
				v.error("parallel.open_branch", ownerPath, fmt.Sprintf("branch %q of %q has a path that does not close at join %q", branch, owner.ID, join.ID), owner.ID, "")
			}
		}
		v.validateParallelBranchMembership(owner, join, branches, scopes, incoming)
		v.validateParallelStateWrites(owner, branches, scopes, outgoing)
		v.recordParallelScope(owner, join, branches, scopes)
	}
}

func (v *validator) exploreParallelBranch(owner, join *Node, branch string, outgoing map[string]map[string][]*Edge) *parallelBranchScope {
	scope := &parallelBranchScope{
		branch: branch, 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 != branch {
				v.error("parallel.cross_wired_branch", fmt.Sprintf("graph.edges[%d].target.port", v.edgeIndexes[edge.ID]), fmt.Sprintf("branch %q reaches join port %q", branch, 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][branch] {
		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 := make([]string, 0, len(outgoing[current]))
		for port := range outgoing[current] {
			ports = append(ports, port)
		}
		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 })
			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) validateParallelBranchMembership(owner, join *Node, branches []string, scopes map[string]*parallelBranchScope, incoming map[string]map[string][]*Edge) {
	memberships := map[string][]string{}
	for _, branch := range branches {
		for nodeID := range scopes[branch].nodes {
			memberships[nodeID] = append(memberships[nodeID], branch)
		}
	}
	for nodeID, memberBranches := range memberships {
		if len(memberBranches) < 2 {
			continue
		}
		if node := v.nodes[nodeID]; node != nil && node.Type == NodeEnd && node.End != nil && node.End.Outcome == EndFailed {
			// A failed End is a token terminal intercepted by the nearest open
			// parallel scope, not executable work shared by the branches.
			continue
		}
		sort.Strings(memberBranches)
		v.error("parallel.branch_overlap", fmt.Sprintf("graph.nodes[%d]", v.nodeIndexes[nodeID]), fmt.Sprintf("node %q is reachable from parallel branches %v before join %q", nodeID, memberBranches, join.ID), nodeID, "")
	}

	for _, branch := range branches {
		edges := append([]*Edge(nil), incoming[join.ID][branch]...)
		sort.Slice(edges, func(i, j int) bool { return edges[i].ID < edges[j].ID })
		for _, edge := range edges {
			valid := edge.Source.NodeID == owner.ID && edge.Source.Port == branch
			if !valid {
				valid = scopes[branch].nodes[edge.Source.NodeID] && len(memberships[edge.Source.NodeID]) == 1
			}
			if !valid {
				v.error("parallel.cross_wired_branch", fmt.Sprintf("graph.edges[%d]", v.edgeIndexes[edge.ID]), fmt.Sprintf("join port %q source %q is not exclusively reachable from owner branch %q", branch, edge.Source.NodeID, branch), owner.ID, edge.ID)
			}
		}
	}
}

func (v *validator) recordParallelScope(owner, join *Node, branches []string, scopes map[string]*parallelBranchScope) {
	activation := "all"
	if owner.Type == NodeSwitch {
		activation = "matched"
	}
	snapshot := parallelScopeAnalysis{
		ownerID: owner.ID, joinID: join.ID, mode: join.Join.Mode, activation: activation,
		branches: make([]parallelBranchAnalysis, 0, len(branches)),
	}
	for _, branch := range branches {
		scope := scopes[branch]
		compiled := parallelBranchAnalysis{
			port: branch, members: []string{}, entryEdges: []string{}, successArrivals: []string{}, failedTerminals: []string{},
		}
		for nodeID := range scope.nodes {
			node := v.nodes[nodeID]
			if node != nil && node.Type != NodeEnd {
				compiled.members = append(compiled.members, nodeID)
			}
		}
		for _, edge := range v.document.Graph.Edges {
			if edge.Source.NodeID == owner.ID && edge.Source.Port == branch {
				compiled.entryEdges = append(compiled.entryEdges, edge.ID)
			}
		}
		for _, boundary := range scope.boundaries {
			compiled.successArrivals = append(compiled.successArrivals, boundary.ID)
		}
		for nodeID := range scope.failedTerminals {
			compiled.failedTerminals = append(compiled.failedTerminals, nodeID)
		}
		sort.Strings(compiled.members)
		sort.Strings(compiled.entryEdges)
		sort.Strings(compiled.successArrivals)
		sort.Strings(compiled.failedTerminals)
		snapshot.branches = append(snapshot.branches, compiled)
	}
	sort.Slice(snapshot.branches, func(i, j int) bool { return snapshot.branches[i].port < snapshot.branches[j].port })
	v.parallelScopes = append(v.parallelScopes, snapshot)
}

func (v *validator) validateParallelStateWrites(owner *Node, branches []string, scopes map[string]*parallelBranchScope, outgoing map[string]map[string][]*Edge) {
	effects := compileDocumentStateEffects(v)
	type branchEffects struct {
		reads  map[string]bool
		writes map[string]bool
	}
	accesses := map[string]branchEffects{}
	for _, branch := range branches {
		access := branchEffects{reads: map[string]bool{}, writes: map[string]bool{}}
		// The owner is outside every concurrent branch, but the mapping on
		// each selected owner port executes while materializing that specific
		// branch activation and therefore belongs to that branch's read set.
		for _, edge := range outgoing[owner.ID][branch] {
			target := v.nodes[edge.Target.NodeID]
			for state := range compileArtifactBindingReadSet(v, edge.Bindings, owner, referenceContext{sourcePort: edge.Source.Port}, v.targetSchema(target, edge.Target.Port)) {
				access.reads[state] = true
			}
		}
		for nodeID := range scopes[branch].nodes {
			for _, state := range effects[nodeID].Reads {
				access.reads[state] = true
			}
			for _, state := range effects[nodeID].Writes {
				access.writes[state] = true
			}
		}
		accesses[branch] = access
	}
	for i, left := range branches {
		for _, right := range branches[i+1:] {
			if parallelBranchesMutuallyExclusive(owner, left, right) {
				continue
			}
			writeConflicts := make([]string, 0)
			for state := range accesses[left].writes {
				if accesses[right].writes[state] {
					writeConflicts = append(writeConflicts, state)
				}
			}
			sort.Strings(writeConflicts)
			writeConflict := map[string]bool{}
			for _, state := range writeConflicts {
				writeConflict[state] = true
				v.error("parallel.state_write_conflict", fmt.Sprintf("graph.nodes[%d]", v.nodeIndexes[owner.ID]), fmt.Sprintf("branches %q and %q can both write state %q", left, right, state), owner.ID, "")
			}

			// A write racing a read is nondeterministic even when there is only
			// one writer. Report each direction explicitly; W/W already carries
			// the stronger diagnostic and is not duplicated as W/R.
			for _, pair := range []struct {
				writer string
				reader string
			}{{left, right}, {right, left}} {
				conflicts := make([]string, 0)
				for state := range accesses[pair.writer].writes {
					if !writeConflict[state] && accesses[pair.reader].reads[state] {
						conflicts = append(conflicts, state)
					}
				}
				sort.Strings(conflicts)
				for _, state := range conflicts {
					v.error("parallel.state_access_conflict", fmt.Sprintf("graph.nodes[%d]", v.nodeIndexes[owner.ID]), fmt.Sprintf("branch %q can write state %q while branch %q reads it", pair.writer, state, pair.reader), owner.ID, "")
				}
			}
		}
	}
}

func parallelBranchesMutuallyExclusive(owner *Node, left, right string) bool {
	if owner == nil || owner.Type != NodeSwitch || owner.Switch == nil || owner.Switch.Mode != SwitchAllMatch {
		return false
	}
	defaultPort := ""
	for _, switchCase := range owner.Switch.Cases {
		if switchCase.IsDefault {
			defaultPort = switchCase.Port
			break
		}
	}
	return defaultPort != "" && (left == defaultPort || right == defaultPort)
}

func (v *validator) reachableNodes() map[string]bool {
	entry := v.document.Graph.Entry.NodeID
	reachable := map[string]bool{}
	if _, exists := v.nodes[entry]; !exists {
		return reachable
	}
	adjacency := v.adjacency(false)
	queue := []string{entry}
	reachable[entry] = true
	for len(queue) != 0 {
		current := queue[0]
		queue = queue[1:]
		for _, next := range adjacency[current] {
			if !reachable[next] {
				reachable[next] = true
				queue = append(queue, next)
			}
		}
	}
	return reachable
}

func (v *validator) dominators(reachable map[string]bool) map[string]map[string]bool {
	entry := v.document.Graph.Entry.NodeID
	predecessors := map[string][]string{}
	for _, edge := range v.document.Graph.Edges {
		if reachable[edge.Source.NodeID] && reachable[edge.Target.NodeID] {
			predecessors[edge.Target.NodeID] = append(predecessors[edge.Target.NodeID], edge.Source.NodeID)
		}
	}
	all := map[string]bool{}
	for nodeID := range reachable {
		all[nodeID] = true
	}
	dominators := map[string]map[string]bool{}
	for nodeID := range reachable {
		if nodeID == entry {
			dominators[nodeID] = map[string]bool{entry: true}
		} else {
			dominators[nodeID] = copySet(all)
		}
	}
	changed := true
	for changed {
		changed = false
		for nodeID := range reachable {
			if nodeID == entry {
				continue
			}
			preds := predecessors[nodeID]
			next := map[string]bool{nodeID: true}
			if len(preds) != 0 {
				intersection := copySet(dominators[preds[0]])
				for _, predecessor := range preds[1:] {
					for candidate := range intersection {
						if !dominators[predecessor][candidate] {
							delete(intersection, candidate)
						}
					}
				}
				for candidate := range intersection {
					next[candidate] = true
				}
			}
			if !equalSet(dominators[nodeID], next) {
				dominators[nodeID] = next
				changed = true
			}
		}
	}
	return dominators
}

func copySet(source map[string]bool) map[string]bool {
	copy := make(map[string]bool, len(source))
	for value := range source {
		copy[value] = true
	}
	return copy
}

func equalSet(left, right map[string]bool) bool {
	if len(left) != len(right) {
		return false
	}
	for value := range left {
		if !right[value] {
			return false
		}
	}
	return true
}