feat(search): opensearch aggregations

Implements terms, range, metric and sub-aggregations for the OpenSearch backend
via a dedicated aggs builder, wiring them through the shared search service.

services/search/pkg/opensearch/backend.go

@@ -16,6 +16,7 @@ import (
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	searchMessage "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
 	searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/aggs"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/convert"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/osu"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
@@ -124,6 +125,7 @@ func (b *Backend) Search(ctx context.Context, sir *searchService.SearchIndexRequ
 					},
 				},
 			},
+			Aggs: aggs.Build(sir.GetAggregations()),
 		},
 	)
 	if err != nil {
@@ -157,9 +159,15 @@ func (b *Backend) Search(ctx context.Context, sir *searchService.SearchIndexRequ
 		matches = append(matches, match)
 	}
 
+	aggResults, err := aggs.Parse(resp.Aggregations, sir.GetAggregations())
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse aggregations: %w", err)
+	}
+
 	return &searchService.SearchIndexResponse{
 		Matches:      matches,
 		TotalMatches: int32(totalMatches),
+		Aggregations: aggResults,
 	}, nil
 }
 

services/search/pkg/opensearch/internal/aggs/aggs.go

@@ -0,0 +1,258 @@
+// Package aggs translates proto aggregation options into the OpenSearch
+// aggregation DSL and parses the response. Internal subpackage so its unit
+// tests skip the parent package's Docker OpenSearch container.
+package aggs
+
+import (
+	"encoding/json"
+	"fmt"
+	"strconv"
+
+	searchsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
+)
+
+// DefaultFacetSize matches the bleve backend: pull a generous bucket count per
+// space, the service layer trims to top N after cross-space merge.
+const DefaultFacetSize = 1000
+
+// Build translates AggregationOptions into the OpenSearch aggregation DSL
+// (terms, range, metric, nested). Entries get an index-derived name so repeated
+// aggs on one field don't collide.
+func Build(opts []*searchsvc.AggregationOption) map[string]any {
+	return buildLevel(opts, "a")
+}
+
+func buildLevel(opts []*searchsvc.AggregationOption, prefix string) map[string]any {
+	if len(opts) == 0 {
+		return nil
+	}
+	aggs := map[string]any{}
+	for i, opt := range opts {
+		name := fmt.Sprintf("%s_%d", prefix, i)
+		if entry := buildOne(opt, name); entry != nil {
+			aggs[name] = entry
+		}
+	}
+	if len(aggs) == 0 {
+		return nil
+	}
+	return aggs
+}
+
+func buildOne(opt *searchsvc.AggregationOption, name string) map[string]any {
+	field := opt.GetField()
+	if mk := opt.GetMetricKind(); mk != searchsvc.MetricKind_METRIC_KIND_UNSPECIFIED {
+		return buildMetric(field, mk)
+	}
+	var entry map[string]any
+	if ranges := rangesOf(opt); len(ranges) > 0 {
+		entry = map[string]any{
+			"range": map[string]any{
+				"field":  field,
+				"ranges": buildRanges(ranges),
+			},
+		}
+	} else {
+		size := int(opt.GetSize())
+		if size <= 0 {
+			size = DefaultFacetSize
+		}
+		entry = map[string]any{
+			"terms": map[string]any{
+				"field": field,
+				"size":  size,
+			},
+		}
+	}
+	if subs := opt.GetSubAggregations(); len(subs) > 0 {
+		if nested := buildLevel(subs, name); nested != nil {
+			entry["aggs"] = nested
+		}
+	}
+	return entry
+}
+
+// buildMetric emits the sum/min/max metric. AVG uses a stats agg to transport
+// (sum, count) for the cross-space merge; the service layer collapses to the average.
+func buildMetric(field string, kind searchsvc.MetricKind) map[string]any {
+	switch kind {
+	case searchsvc.MetricKind_METRIC_KIND_SUM:
+		return map[string]any{"sum": map[string]any{"field": field}}
+	case searchsvc.MetricKind_METRIC_KIND_MIN:
+		return map[string]any{"min": map[string]any{"field": field}}
+	case searchsvc.MetricKind_METRIC_KIND_MAX:
+		return map[string]any{"max": map[string]any{"field": field}}
+	case searchsvc.MetricKind_METRIC_KIND_AVG:
+		return map[string]any{"stats": map[string]any{"field": field}}
+	}
+	return nil
+}
+
+func rangesOf(opt *searchsvc.AggregationOption) []*searchsvc.BucketRange {
+	bd := opt.GetBucketDefinition()
+	if bd == nil {
+		return nil
+	}
+	return bd.GetRanges()
+}
+
+func buildRanges(ranges []*searchsvc.BucketRange) []map[string]any {
+	out := make([]map[string]any, 0, len(ranges))
+	for _, r := range ranges {
+		entry := map[string]any{
+			"key": RangeKey(r),
+		}
+		if v, err := strconv.ParseFloat(r.GetFrom(), 64); err == nil {
+			entry["from"] = v
+		}
+		if v, err := strconv.ParseFloat(r.GetTo(), 64); err == nil {
+			entry["to"] = v
+		}
+		out = append(out, entry)
+	}
+	return out
+}
+
+// RangeKey mirrors the bleve backend so cross-space merging keys match.
+func RangeKey(r *searchsvc.BucketRange) string {
+	return r.GetFrom() + "-" + r.GetTo()
+}
+
+// Parse converts the response aggregations block into proto results, preserving
+// request order and recursing into sub-aggregations. Empty input yields
+// (nil, nil); invalid JSON yields an error.
+func Parse(raw json.RawMessage, opts []*searchsvc.AggregationOption) ([]*searchsvc.AggregationResult, error) {
+	if len(raw) == 0 || len(opts) == 0 {
+		return nil, nil
+	}
+	node, err := parseNode(raw)
+	if err != nil {
+		return nil, err
+	}
+	return parseLevel(node, opts, "a"), nil
+}
+
+// aggNode is a lazily-decoded cursor over one level of the aggs response.
+type aggNode map[string]json.RawMessage
+
+func parseNode(raw json.RawMessage) (aggNode, error) {
+	var m aggNode
+	if err := json.Unmarshal(raw, &m); err != nil {
+		return nil, fmt.Errorf("decode opensearch aggregations: %w", err)
+	}
+	return m, nil
+}
+
+func parseLevel(node aggNode, opts []*searchsvc.AggregationOption, prefix string) []*searchsvc.AggregationResult {
+	out := make([]*searchsvc.AggregationResult, 0, len(opts))
+	for i, opt := range opts {
+		name := fmt.Sprintf("%s_%d", prefix, i)
+		raw, ok := node[name]
+		if !ok {
+			continue
+		}
+		if res := parseOne(raw, opt, name); res != nil {
+			out = append(out, res)
+		}
+	}
+	return out
+}
+
+func parseOne(raw json.RawMessage, opt *searchsvc.AggregationOption, name string) *searchsvc.AggregationResult {
+	field := opt.GetField()
+	if mk := opt.GetMetricKind(); mk != searchsvc.MetricKind_METRIC_KIND_UNSPECIFIED {
+		return parseMetric(raw, field, mk)
+	}
+	var body struct {
+		Buckets []json.RawMessage `json:"buckets"`
+	}
+	if err := json.Unmarshal(raw, &body); err != nil {
+		return nil
+	}
+	buckets := make([]*searchsvc.Bucket, 0, len(body.Buckets))
+	for _, b := range body.Buckets {
+		if bucket := parseBucket(b, opt.GetSubAggregations(), name); bucket != nil {
+			buckets = append(buckets, bucket)
+		}
+	}
+	return &searchsvc.AggregationResult{
+		Field:   field,
+		Buckets: buckets,
+	}
+}
+
+func parseBucket(raw json.RawMessage, subs []*searchsvc.AggregationOption, prefix string) *searchsvc.Bucket {
+	var head struct {
+		Key      any   `json:"key"`
+		DocCount int64 `json:"doc_count"`
+	}
+	if err := json.Unmarshal(raw, &head); err != nil {
+		return nil
+	}
+	b := &searchsvc.Bucket{
+		Key:   bucketKeyToString(head.Key),
+		Count: head.DocCount,
+	}
+	if len(subs) > 0 {
+		node, err := parseNode(raw)
+		if err == nil {
+			b.SubAggregations = parseLevel(node, subs, prefix)
+		}
+	}
+	return b
+}
+
+func parseMetric(raw json.RawMessage, field string, kind searchsvc.MetricKind) *searchsvc.AggregationResult {
+	switch kind {
+	case searchsvc.MetricKind_METRIC_KIND_SUM,
+		searchsvc.MetricKind_METRIC_KIND_MIN,
+		searchsvc.MetricKind_METRIC_KIND_MAX:
+		var body struct {
+			Value *float64 `json:"value"`
+		}
+		if err := json.Unmarshal(raw, &body); err != nil {
+			return nil
+		}
+		res := &searchsvc.AggregationResult{
+			Field:      field,
+			MetricKind: kind,
+		}
+		if body.Value != nil {
+			res.Value = *body.Value
+		}
+		return res
+	case searchsvc.MetricKind_METRIC_KIND_AVG:
+		var body struct {
+			Sum   float64 `json:"sum"`
+			Count int64   `json:"count"`
+		}
+		if err := json.Unmarshal(raw, &body); err != nil {
+			return nil
+		}
+		return &searchsvc.AggregationResult{
+			Field:      field,
+			Sum:        body.Sum,
+			Count:      body.Count,
+			MetricKind: kind,
+		}
+	}
+	return nil
+}
+
+// bucketKeyToString normalises a response key to a string (terms are strings,
+// ranges use our "from-to" key, numeric terms come back as JSON numbers).
+func bucketKeyToString(v any) string {
+	switch x := v.(type) {
+	case string:
+		return x
+	case float64:
+		// format without trailing zeros so keys match filter values
+		return strconv.FormatFloat(x, 'f', -1, 64)
+	case bool:
+		return strconv.FormatBool(x)
+	case nil:
+		return ""
+	default:
+		return ""
+	}
+}

services/search/pkg/opensearch/internal/aggs/aggs_suite_test.go

@@ -0,0 +1,13 @@
+package aggs_test
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestAggs(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Aggs Suite")
+}

services/search/pkg/opensearch/internal/aggs/aggs_test.go

@@ -0,0 +1,227 @@
+package aggs_test
+
+import (
+	"encoding/json"
+	"fmt"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	searchsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/aggs"
+)
+
+var _ = Describe("Build", func() {
+	It("builds a terms aggregation", func() {
+		res := aggs.Build([]*searchsvc.AggregationOption{
+			{Field: "audio.artist", Size: 10},
+		})
+		Expect(res).ToNot(BeNil())
+		entry, ok := res["a_0"].(map[string]any)
+		Expect(ok).To(BeTrue())
+		terms, ok := entry["terms"].(map[string]any)
+		Expect(ok).To(BeTrue())
+		Expect(terms["field"]).To(Equal("audio.artist"))
+		Expect(terms["size"]).To(Equal(10))
+	})
+
+	It("builds a range aggregation with open-ended bounds", func() {
+		res := aggs.Build([]*searchsvc.AggregationOption{{
+			Field: "audio.year",
+			BucketDefinition: &searchsvc.BucketDefinition{
+				Ranges: []*searchsvc.BucketRange{
+					{From: "1970", To: "1980"},
+					{To: "1970"},
+					{From: "2020"},
+				},
+			},
+		}})
+		r := res["a_0"].(map[string]any)["range"].(map[string]any)
+		Expect(r["field"]).To(Equal("audio.year"))
+		ranges := r["ranges"].([]map[string]any)
+		Expect(ranges).To(HaveLen(3))
+		Expect(ranges[0]).To(SatisfyAll(
+			HaveKeyWithValue("key", "1970-1980"),
+			HaveKeyWithValue("from", 1970.0),
+			HaveKeyWithValue("to", 1980.0),
+		))
+		Expect(ranges[1]).ToNot(HaveKey("from")) // open lower bound
+		Expect(ranges[2]).ToNot(HaveKey("to"))   // open upper bound
+	})
+
+	DescribeTable("builds single-value metric aggregations",
+		func(kind searchsvc.MetricKind, esKind string) {
+			res := aggs.Build([]*searchsvc.AggregationOption{
+				{Field: "audio.duration", MetricKind: kind},
+			})
+			body, ok := res["a_0"].(map[string]any)[esKind].(map[string]any)
+			Expect(ok).To(BeTrue())
+			Expect(body["field"]).To(Equal("audio.duration"))
+		},
+		Entry("sum", searchsvc.MetricKind_METRIC_KIND_SUM, "sum"),
+		Entry("min", searchsvc.MetricKind_METRIC_KIND_MIN, "min"),
+		Entry("max", searchsvc.MetricKind_METRIC_KIND_MAX, "max"),
+	)
+
+	It("uses a stats aggregation for AVG", func() {
+		res := aggs.Build([]*searchsvc.AggregationOption{
+			{Field: "audio.duration", MetricKind: searchsvc.MetricKind_METRIC_KIND_AVG},
+		})
+		stats, ok := res["a_0"].(map[string]any)["stats"].(map[string]any)
+		Expect(ok).To(BeTrue())
+		Expect(stats["field"]).To(Equal("audio.duration"))
+	})
+
+	It("nests sub-aggregations under their parent bucket", func() {
+		res := aggs.Build([]*searchsvc.AggregationOption{{
+			Field: "audio.artist", Size: 5,
+			SubAggregations: []*searchsvc.AggregationOption{{
+				Field: "audio.album", Size: 7,
+				SubAggregations: []*searchsvc.AggregationOption{
+					{Field: "audio.duration", MetricKind: searchsvc.MetricKind_METRIC_KIND_SUM},
+				},
+			}},
+		}})
+		album := res["a_0"].(map[string]any)["aggs"].(map[string]any)["a_0_0"].(map[string]any)
+		albumTerms := album["terms"].(map[string]any)
+		Expect(albumTerms["field"]).To(Equal("audio.album"))
+		Expect(albumTerms["size"]).To(Equal(7))
+		metric := album["aggs"].(map[string]any)["a_0_0_0"].(map[string]any)
+		Expect(metric["sum"].(map[string]any)["field"]).To(Equal("audio.duration"))
+	})
+})
+
+var _ = Describe("Parse", func() {
+	It("parses flat term and range buckets, stringifying numeric keys", func() {
+		raw := json.RawMessage(`{
+			"a_0": {"buckets": [
+				{"key": "Pink Floyd", "doc_count": 42},
+				{"key": "Motörhead", "doc_count": 35}
+			]},
+			"a_1": {"buckets": [
+				{"key": "1970-1980", "from": 1970.0, "to": 1980.0, "doc_count": 12}
+			]},
+			"a_2": {"buckets": [
+				{"key": 9, "doc_count": 3}
+			]}
+		}`)
+		out, err := aggs.Parse(raw, []*searchsvc.AggregationOption{
+			{Field: "audio.artist"},
+			{Field: "audio.year"},
+			{Field: "audio.track"},
+		})
+		Expect(err).ToNot(HaveOccurred())
+		Expect(out).To(HaveLen(3))
+
+		Expect(out[0].Field).To(Equal("audio.artist"))
+		Expect(out[0].Buckets).To(HaveLen(2))
+		Expect(out[0].Buckets[0].Key).To(Equal("Pink Floyd"))
+		Expect(out[0].Buckets[0].Count).To(Equal(int64(42)))
+
+		Expect(out[1].Buckets[0].Key).To(Equal("1970-1980"))
+		Expect(out[1].Buckets[0].Count).To(Equal(int64(12)))
+
+		// numeric term key stringified without trailing zeros
+		Expect(out[2].Buckets[0].Key).To(Equal("9"))
+	})
+
+	It("parses nested buckets carrying a metric", func() {
+		raw := json.RawMessage(`{
+			"a_0": {"buckets": [{
+				"key": "Iron Maiden", "doc_count": 300,
+				"a_0_0": {"buckets": [
+					{"key": "The Number of the Beast", "doc_count": 8, "a_0_0_0": {"value": 2756000.0}},
+					{"key": "Powerslave", "doc_count": 8, "a_0_0_0": {"value": 3061000.0}}
+				]}
+			}]}
+		}`)
+		out, err := aggs.Parse(raw, []*searchsvc.AggregationOption{{
+			Field: "audio.artist",
+			SubAggregations: []*searchsvc.AggregationOption{{
+				Field: "audio.album",
+				SubAggregations: []*searchsvc.AggregationOption{
+					{Field: "audio.duration", MetricKind: searchsvc.MetricKind_METRIC_KIND_SUM},
+				},
+			}},
+		}})
+		Expect(err).ToNot(HaveOccurred())
+		Expect(out).To(HaveLen(1))
+		Expect(out[0].Field).To(Equal("audio.artist"))
+		Expect(out[0].Buckets).To(HaveLen(1))
+
+		artistBucket := out[0].Buckets[0]
+		Expect(artistBucket.Key).To(Equal("Iron Maiden"))
+		Expect(artistBucket.Count).To(Equal(int64(300)))
+		Expect(artistBucket.SubAggregations).To(HaveLen(1))
+
+		albumAgg := artistBucket.SubAggregations[0]
+		Expect(albumAgg.Field).To(Equal("audio.album"))
+		Expect(albumAgg.Buckets).To(HaveLen(2))
+
+		nob := albumAgg.Buckets[0]
+		Expect(nob.Key).To(Equal("The Number of the Beast"))
+		Expect(nob.Count).To(Equal(int64(8)))
+		Expect(nob.SubAggregations).To(HaveLen(1))
+
+		metric := nob.SubAggregations[0]
+		Expect(metric.Field).To(Equal("audio.duration"))
+		Expect(metric.MetricKind).To(Equal(searchsvc.MetricKind_METRIC_KIND_SUM))
+		Expect(metric.Value).To(Equal(2756000.0))
+	})
+
+	DescribeTable("parses single-value metrics",
+		func(kind searchsvc.MetricKind, value float64) {
+			raw := json.RawMessage(fmt.Sprintf(`{"a_0": {"value": %g}}`, value))
+			out, err := aggs.Parse(raw, []*searchsvc.AggregationOption{
+				{Field: "audio.duration", MetricKind: kind},
+			})
+			Expect(err).ToNot(HaveOccurred())
+			Expect(out).To(HaveLen(1))
+			Expect(out[0].MetricKind).To(Equal(kind))
+			Expect(out[0].Value).To(Equal(value))
+		},
+		Entry("sum", searchsvc.MetricKind_METRIC_KIND_SUM, 1234.5),
+		Entry("min", searchsvc.MetricKind_METRIC_KIND_MIN, 10.0),
+		Entry("max", searchsvc.MetricKind_METRIC_KIND_MAX, 99.0),
+	)
+
+	It("decodes a null metric value to zero", func() {
+		// OpenSearch returns value: null when a metric has no matching docs.
+		out, err := aggs.Parse(json.RawMessage(`{"a_0": {"value": null}}`),
+			[]*searchsvc.AggregationOption{{Field: "audio.duration", MetricKind: searchsvc.MetricKind_METRIC_KIND_SUM}})
+		Expect(err).ToNot(HaveOccurred())
+		Expect(out).To(HaveLen(1))
+		Expect(out[0].Value).To(BeZero())
+		Expect(out[0].MetricKind).To(Equal(searchsvc.MetricKind_METRIC_KIND_SUM))
+	})
+
+	It("carries avg transport (sum + count) from a stats response", func() {
+		raw := json.RawMessage(`{
+			"a_0": {"count": 100, "min": 30000.0, "max": 500000.0, "avg": 245000.0, "sum": 24500000.0}
+		}`)
+		out, err := aggs.Parse(raw, []*searchsvc.AggregationOption{
+			{Field: "audio.duration", MetricKind: searchsvc.MetricKind_METRIC_KIND_AVG},
+		})
+		Expect(err).ToNot(HaveOccurred())
+		Expect(out).To(HaveLen(1))
+		Expect(out[0].MetricKind).To(Equal(searchsvc.MetricKind_METRIC_KIND_AVG))
+		Expect(out[0].Sum).To(Equal(24500000.0))
+		Expect(out[0].Count).To(Equal(int64(100)))
+	})
+
+	It("returns nil for empty raw or empty options", func() {
+		got, err := aggs.Parse(nil, []*searchsvc.AggregationOption{{Field: "x"}})
+		Expect(err).ToNot(HaveOccurred())
+		Expect(got).To(BeNil())
+
+		got, err = aggs.Parse(json.RawMessage(`{}`), nil)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(got).To(BeNil())
+	})
+
+	It("errors on malformed json and returns no result", func() {
+		got, err := aggs.Parse(json.RawMessage(`not-json`), []*searchsvc.AggregationOption{{Field: "x"}})
+		Expect(err).To(HaveOccurred())
+		Expect(got).To(BeNil())
+	})
+})

services/search/pkg/opensearch/internal/osu/request.go

@@ -95,6 +95,7 @@ func BuildSearchReq(req *opensearchgoAPI.SearchReq, q Builder, p ...SearchBodyPa
 
 type SearchBodyParams struct {
 	Highlight *BodyParamHighlight `json:"highlight,omitempty"`
+	Aggs      map[string]any      `json:"aggs,omitempty"`
 }
 
 //----------------------------------------------------------------------------//