// Neighbor-counting and common-neighbor analytics: the neighbor histogram,
// the pairwise common-neighborhood set, and the masked A^2 / link-
// prediction primitive (per-source distinct two-hop endpoint counts,
// folded in parallel). Split from kernels.go, which holds the
// roots-via / functional-relationship kernels.
package chickpeas

import (
	

	
	
	
)

// NeighborCounts is the histogram of neighbor nodes reached from sources
// via the m-matched rels in dir (how many of the sources point to each
// neighbor). Counts into pooled generation-stamped dense scratch.
func ( *Snapshot) ( []NodeID,  Direction,  RelMatch) map[NodeID]int {
	 := int(.CSRIDSpace())
	 := takeScratch()
	defer scratchPool.Put()
	var  []NodeID
	for ,  := range  {
		for  := range .NeighborsMatch(, , ) {
			if .gen[] != .cur {
				.gen[] = .cur
				.dist[] = 0
				 = append(, )
			}
			.dist[]++
		}
	}
	 := make(map[NodeID]int, len())
	for ,  := range  {
		[] = int(.dist[])
	}
	return 
}

// CommonNeighbors is N(a) ∩ N(b) over the m-matched rels in dir (Both
// gives the undirected common neighborhood) -- the link-prediction
// primitive, returned as a set so it composes with And/Or/AndNot.
func ( *Snapshot) (,  NodeID,  Direction,  RelMatch) *nodeset.Set {
	,  := roaring.New(), roaring.New()
	for  := range .NeighborsMatch(, , ) {
		.Add()
	}
	for  := range .NeighborsMatch(, , ) {
		.Add()
	}
	return nodeset.FromBitmap(roaring.And(, ))
}

// CommonNeighborCount is one (source, target, count) triple of
// CommonNeighborCounts.
type CommonNeighborCount struct {
	Source, Target NodeID
	Count          uint64
}

// pairScratch is the per-worker scratch behind CommonNeighborCounts: gen-
// stamped counts plus two dedup stamps (per-source mids, per-mid targets).
type pairScratch struct {
	val          []uint64
	gen, seenMid []uint32
	tSeen        []uint32
	cur, tCur    uint32
	touched      []NodeID
}

var pairScratchPool = sync.Pool{New: func() any { return &pairScratch{} }}

// CommonNeighborCounts is the masked A^2 / link-prediction primitive: for
// each s in sources, count the DISTINCT mid nodes on two-hop paths
// s -[m]- mid -[m]- t whose endpoint t is in targets, emitting one
// (s, t, count) per reached pair (count > 0 only). count is |N(s) ∩ N(t)|
// over de-duplicated neighborhoods -- exact even when the rel is stored
// bidirectionally and traversed Both -- not a walk multiplicity. (s, s)
// self-pairs ARE emitted; callers wanting distinct endpoints filter them.
// Sources are processed in parallel (chunk results concatenated in source
// order, so output order is deterministic); counting scatters from each
// source through its mids (a "push"), far cheaper than per-pair set
// intersections when few pairs share a neighbor.
// cncChunk is the pair accumulator's slab size: fixed chunks replace the
// per-worker doubling ladder (each doubling discarded its predecessor,
// dominating the primitive's allocation profile), and the final assembly
// is one exact-size copy.
const cncChunk = 4096

// cncAcc is one worker's accumulation state: the worker-held scratch
// (one pool round-trip per worker, not per source) and the ordered pair
// slabs.
type cncAcc struct {
	sc   *pairScratch
	full [][]CommonNeighborCount
	cur  []CommonNeighborCount
}

// seal closes the open slab into the ordered list.
func ( *cncAcc) () {
	if len(.cur) > 0 {
		.full = append(.full, .cur)
		.cur = nil
	}
}

func ( *Snapshot) ( []NodeID,  Direction,  RelMatch,  *nodeset.Set) []CommonNeighborCount {
	 := int(.CSRIDSpace())
	 := parallel.Fold(len(),
		func() *cncAcc { return &cncAcc{sc: pairScratchPool.Get().(*pairScratch)} },
		func( *cncAcc,  int) *cncAcc {
			 := []
			 := .sc
			if len(.val) <  {
				.val = make([]uint64, )
				.gen = make([]uint32, )
				.seenMid = make([]uint32, )
				.tSeen = make([]uint32, )
				.cur, .tCur = 0, 0
			}
			.cur++
			if .cur == 0 {
				clear(.gen)
				clear(.seenMid)
				.cur = 1
			}
			 := .cur
			.touched = .touched[:0]
			for  := range .NeighborsMatch(, , ) {
				// First-hop dedup: process each distinct mid once.
				if .seenMid[] ==  {
					continue
				}
				.seenMid[] = 
				.tCur++
				if .tCur == 0 {
					clear(.tSeen)
					.tCur = 1
				}
				 := .tCur
				for  := range .NeighborsMatch(, , ) {
					if !.Contains() {
						continue
					}
					// Second-hop dedup: bump each distinct endpoint once per mid.
					if .tSeen[] ==  {
						continue
					}
					.tSeen[] = 
					if .gen[] !=  {
						.gen[] = 
						.val[] = 0
						.touched = append(.touched, )
					}
					.val[]++
				}
			}
			for ,  := range .touched {
				if len(.cur) == cncChunk {
					.full = append(.full, .cur)
					.cur = make([]CommonNeighborCount, 0, cncChunk)
				}
				if .cur == nil {
					.cur = make([]CommonNeighborCount, 0, cncChunk)
				}
				.cur = append(.cur, CommonNeighborCount{Source: , Target: , Count: .val[]})
			}
			return 
		},
		// Merge keeps slab order (ascending worker ranges): the left side's
		// open slab seals before the right side's slabs append, and the
		// right side's scratch returns to the pool.
		func(,  *cncAcc) *cncAcc {
			.seal()
			.seal()
			.full = append(.full, .full...)
			if .sc != nil {
				pairScratchPool.Put(.sc)
				.sc = nil
			}
			return 
		})
	.seal()
	if .sc != nil {
		pairScratchPool.Put(.sc)
		.sc = nil
	}
	 := 0
	for ,  := range .full {
		 += len()
	}
	 := make([]CommonNeighborCount, 0, )
	for ,  := range .full {
		 = append(, ...)
	}
	return 
}