databroker: add support for querying the databroker (#1443)

* databroker: add support for querying the databroker

* remove query method, use getall so encryption works

* add test

* return early

internal/databroker/server.go

@@ -9,6 +9,7 @@ import (
 	"io/ioutil"
 	"reflect"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -220,6 +221,43 @@ func (srv *Server) GetAll(ctx context.Context, req *databroker.GetAllRequest) (*
 	}, nil
 }
 
+// Query queries for records.
+func (srv *Server) Query(ctx context.Context, req *databroker.QueryRequest) (*databroker.QueryResponse, error) {
+	_, span := trace.StartSpan(ctx, "databroker.grpc.Query")
+	defer span.End()
+	srv.log.Info().
+		Str("type", req.GetType()).
+		Str("query", req.GetQuery()).
+		Int64("offset", req.GetOffset()).
+		Int64("limit", req.GetLimit()).
+		Msg("query")
+
+	query := strings.ToLower(req.GetQuery())
+
+	db, _, err := srv.getDB(req.GetType(), true)
+	if err != nil {
+		return nil, err
+	}
+
+	all, err := db.GetAll(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	var filtered []*databroker.Record
+	for _, record := range all {
+		if record.DeletedAt == nil && storage.MatchAny(record.GetData(), query) {
+			filtered = append(filtered, record)
+		}
+	}
+
+	records, totalCount := databroker.ApplyOffsetAndLimit(filtered, int(req.GetOffset()), int(req.GetLimit()))
+	return &databroker.QueryResponse{
+		Records:    records,
+		TotalCount: int64(totalCount),
+	}, nil
+}
+
 // Set updates a record in the in-memory list, or adds a new one.
 func (srv *Server) Set(ctx context.Context, req *databroker.SetRequest) (*databroker.SetResponse, error) {
 	_, span := trace.StartSpan(ctx, "databroker.grpc.Set")