core/utils: add BoundedQueue methods, collection helpers, and fix Bat…

core/utils/collection.go

@@ -1,11 +1,16 @@
 package utils
 
-import "errors"
+import (
+	"errors"
+	"maps"
+	"slices"
+)
 
-// BatchSplit splits an slices into an slices of slicess with a maximum length
+// BatchSplit splits a slice into a slice of slices with a maximum length.
+// Returns an error if max is less than or equal to zero.
 func BatchSplit[T any](list []T, max int) (out [][]T, err error) {
-	if max == 0 {
-		return out, errors.New("max batch length cannot be 0")
+	if max <= 0 {
+		return out, errors.New("max batch length must be greater than 0")
 	}
 
 	// batch list into no more than max each
@@ -17,3 +22,73 @@ func BatchSplit[T any](list []T, max int) (out [][]T, err error) {
 	out = append(out, list) // append remaining to list (slice len < max)
 	return out, nil
 }
+
+// Flatten takes a slice of slices and returns a single concatenated slice.
+func Flatten[T any](lists [][]T) []T {
+	var total int
+	for _, l := range lists {
+		total += len(l)
+	}
+	result := make([]T, 0, total)
+	for _, l := range lists {
+		result = append(result, l...)
+	}
+	return result
+}
+
+// UniqueValues returns the unique values from a map, in sorted order if the
+// values are comparable. The caller receives a new slice.
+func UniqueValues[K comparable, V comparable](m map[K]V) []V {
+	seen := make(map[V]struct{}, len(m))
+	result := make([]V, 0, len(m))
+	for _, v := range m {
+		if _, ok := seen[v]; !ok {
+			seen[v] = struct{}{}
+			result = append(result, v)
+		}
+	}
+	return result
+}
+
+// MergeMaps merges multiple maps into one. Later maps take precedence for
+// duplicate keys.
+func MergeMaps[K comparable, V any](ms ...map[K]V) map[K]V {
+	out := make(map[K]V)
+	for _, m := range ms {
+		maps.Copy(out, m)
+	}
+	return out
+}
+
+// FilterSlice returns a new slice containing only the elements for which the
+// predicate returns true.
+func FilterSlice[T any](s []T, pred func(T) bool) []T {
+	result := make([]T, 0, len(s))
+	for _, v := range s {
+		if pred(v) {
+			result = append(result, v)
+		}
+	}
+	return slices.Clip(result)
+}
+
+// MapSlice applies a function to each element of a slice and returns the results.
+func MapSlice[T any, U any](s []T, fn func(T) U) []U {
+	result := make([]U, len(s))
+	for i, v := range s {
+		result[i] = fn(v)
+	}
+	return result
+}
+
+// ContainsDuplicate returns true if the slice contains duplicate elements.
+func ContainsDuplicate[T comparable](s []T) bool {
+	seen := make(map[T]struct{}, len(s))
+	for _, v := range s {
+		if _, ok := seen[v]; ok {
+			return true
+		}
+		seen[v] = struct{}{}
+	}
+	return false
+}

core/utils/collection_test.go

@@ -6,6 +6,127 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+func TestFlatten(t *testing.T) {
+	t.Parallel()
+
+	t.Run("multiple slices", func(t *testing.T) {
+		t.Parallel()
+		result := Flatten([][]int{{1, 2}, {3, 4}, {5}})
+		assert.Equal(t, []int{1, 2, 3, 4, 5}, result)
+	})
+
+	t.Run("empty input", func(t *testing.T) {
+		t.Parallel()
+		result := Flatten[int](nil)
+		assert.Empty(t, result)
+	})
+
+	t.Run("slices with empty sub-slices", func(t *testing.T) {
+		t.Parallel()
+		result := Flatten([][]int{{1}, {}, {2, 3}, {}})
+		assert.Equal(t, []int{1, 2, 3}, result)
+	})
+}
+
+func TestUniqueValues(t *testing.T) {
+	t.Parallel()
+
+	t.Run("with duplicates", func(t *testing.T) {
+		t.Parallel()
+		m := map[string]int{"a": 1, "b": 2, "c": 1, "d": 3}
+		result := UniqueValues(m)
+		assert.Len(t, result, 3)
+		assert.ElementsMatch(t, []int{1, 2, 3}, result)
+	})
+
+	t.Run("empty map", func(t *testing.T) {
+		t.Parallel()
+		result := UniqueValues(map[string]int{})
+		assert.Empty(t, result)
+	})
+}
+
+func TestMergeMaps(t *testing.T) {
+	t.Parallel()
+
+	t.Run("overlapping keys", func(t *testing.T) {
+		t.Parallel()
+		a := map[string]int{"x": 1, "y": 2}
+		b := map[string]int{"y": 3, "z": 4}
+		result := MergeMaps(a, b)
+		assert.Equal(t, map[string]int{"x": 1, "y": 3, "z": 4}, result)
+	})
+
+	t.Run("no maps", func(t *testing.T) {
+		t.Parallel()
+		result := MergeMaps[string, int]()
+		assert.Empty(t, result)
+	})
+
+	t.Run("single map", func(t *testing.T) {
+		t.Parallel()
+		m := map[string]int{"a": 1}
+		result := MergeMaps(m)
+		assert.Equal(t, map[string]int{"a": 1}, result)
+	})
+}
+
+func TestFilterSlice(t *testing.T) {
+	t.Parallel()
+
+	t.Run("filter evens", func(t *testing.T) {
+		t.Parallel()
+		result := FilterSlice([]int{1, 2, 3, 4, 5, 6}, func(n int) bool { return n%2 == 0 })
+		assert.Equal(t, []int{2, 4, 6}, result)
+	})
+
+	t.Run("empty slice", func(t *testing.T) {
+		t.Parallel()
+		result := FilterSlice([]int{}, func(n int) bool { return true })
+		assert.Empty(t, result)
+	})
+
+	t.Run("none match", func(t *testing.T) {
+		t.Parallel()
+		result := FilterSlice([]int{1, 3, 5}, func(n int) bool { return n%2 == 0 })
+		assert.Empty(t, result)
+	})
+}
+
+func TestMapSlice(t *testing.T) {
+	t.Parallel()
+
+	t.Run("double values", func(t *testing.T) {
+		t.Parallel()
+		result := MapSlice([]int{1, 2, 3}, func(n int) int { return n * 2 })
+		assert.Equal(t, []int{2, 4, 6}, result)
+	})
+
+	t.Run("int to string", func(t *testing.T) {
+		t.Parallel()
+		result := MapSlice([]int{1, 2}, func(n int) string {
+			return string(rune('a' + n - 1))
+		})
+		assert.Equal(t, []string{"a", "b"}, result)
+	})
+
+	t.Run("empty", func(t *testing.T) {
+		t.Parallel()
+		result := MapSlice([]int{}, func(n int) int { return n })
+		assert.Empty(t, result)
+	})
+}
+
+func TestContainsDuplicate(t *testing.T) {
+	t.Parallel()
+
+	assert.True(t, ContainsDuplicate([]int{1, 2, 3, 2}))
+	assert.False(t, ContainsDuplicate([]int{1, 2, 3}))
+	assert.False(t, ContainsDuplicate([]int{}))
+	assert.False(t, ContainsDuplicate([]int{1}))
+	assert.True(t, ContainsDuplicate([]string{"a", "b", "a"}))
+}
+
 func TestBatchSplit(t *testing.T) {
 	list := []int{}
 	for i := range 100 {
@@ -28,6 +149,7 @@ func TestBatchSplit(t *testing.T) {
 		{"max=len+1", list, len(list) + 1, 1, len(list), false}, // max exceeds len of list
 		{"zero-list", []int{}, 1, 1, 0, false},                  // zero length list
 		{"zero-max", list, 0, 0, 0, true},                       // zero as max input
+		{"negative-max", list, -1, 0, 0, true},                  // negative max input
 	}
 
 	for _, r := range runs {

core/utils/utils.go

@@ -174,11 +174,47 @@ type BoundedQueue[T any] struct {
 	mu       sync.RWMutex
 }
 
-// NewBoundedQueue creates a new BoundedQueue instance
+// NewBoundedQueue creates a new BoundedQueue instance.
+// The capacity must be greater than zero; a non-positive capacity will be
+// clamped to 1 to avoid silent misuse.
 func NewBoundedQueue[T any](capacity int) *BoundedQueue[T] {
-	var bq BoundedQueue[T]
-	bq.capacity = capacity
-	return &bq
+	if capacity <= 0 {
+		capacity = 1
+	}
+	return &BoundedQueue[T]{capacity: capacity}
+}
+
+// Len returns the current number of items in the queue.
+func (q *BoundedQueue[T]) Len() int {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
+	return len(q.items)
+}
+
+// Peek returns the first item without removing it.
+// The second return value indicates whether an item was available.
+func (q *BoundedQueue[T]) Peek() (T, bool) {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
+	if len(q.items) == 0 {
+		var zero T
+		return zero, false
+	}
+	return q.items[0], true
+}
+
+// Clear removes all items from the queue.
+func (q *BoundedQueue[T]) Clear() {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	q.items = nil
+}
+
+// Cap returns the maximum capacity of the queue.
+func (q *BoundedQueue[T]) Cap() int {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
+	return q.capacity
 }
 
 // Add appends items to a BoundedQueue
@@ -287,6 +323,18 @@ func (q *BoundedPriorityQueue[T]) Empty() bool {
 	return true
 }
 
+// Len returns the total number of items across all priority sub-queues.
+func (q *BoundedPriorityQueue[T]) Len() int {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
+
+	total := 0
+	for _, priority := range q.priorities {
+		total += q.queues[priority].Len()
+	}
+	return total
+}
+
 // TickerBase is an interface for pausable tickers.
 type TickerBase interface {
 	Resume()