feat: Add storage adapter search method

* Add search/lucene parser
* Extend sql adapter with search method
* Extend dynamodb adapter with search method
* Extend memory adapter with search method

Author: Lutherwaves

storage/dynamodb.go

@@ -17,6 +17,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
 
 	"github.com/tink3rlabs/magic/logger"
+	"github.com/tink3rlabs/magic/storage/search/lucene"
 )
 
 type DynamoDBAdapter struct {
@@ -187,6 +188,81 @@ func (s *DynamoDBAdapter) List(dest any, sortKey string, filter map[string]any,
 
 	return nextToken, nil
 }
+func (s *DynamoDBAdapter) Search(dest any, sortKey string, filter string, limit int, cursor string) (string, error) {
+	slog.Debug(fmt.Sprintf(`Searching in DynamoDB with filter: %s`, filter))
+
+	// Get model type from destination slice
+	destType := reflect.TypeOf(dest).Elem().Elem()
+	model := reflect.New(destType).Elem().Interface()
+
+	// Create Lucene parser from model type
+	parser, err := lucene.NewParserFromType(model)
+	if err != nil {
+		slog.Error(`There was an error creating parser from model`)
+		return "", fmt.Errorf("failed to create parser: %w", err)
+	}
+
+	// Parse Lucene query to DynamoDB PartiQL
+	whereClause, params, err := parser.ParseToDynamoDBPartiQL(filter)
+	if err != nil {
+		slog.Error(`There was an error parsing filter to DynamoDB PartiQL`)
+		return "", fmt.Errorf("invalid filter: %w", err)
+	}
+	slog.Debug(fmt.Sprintf(`Where clause: %s`, whereClause))
+
+	// Build the full query
+	query := fmt.Sprintf(`SELECT * FROM "%v"`, s.getTableName(dest))
+	if whereClause != "" {
+		query = query + fmt.Sprintf(` WHERE %s`, whereClause)
+	}
+
+	// Add sorting if sortKey is provided
+	if sortKey != "" {
+		query = query + fmt.Sprintf(` ORDER BY %s`, sortKey)
+	}
+
+	// Prepare the input for the ExecuteStatement operation
+	input := dynamodb.ExecuteStatementInput{
+		Statement:  aws.String(query),
+		Parameters: params,
+		Limit:      aws.Int32(int32(limit + 1)), // Get one extra item for cursor
+	}
+
+	// Set the cursor if provided
+	if cursor != "" {
+		input.NextToken = aws.String(cursor)
+	}
+
+	// Execute the query
+	response, err := s.DB.ExecuteStatement(context.TODO(), &input)
+	if err != nil {
+		slog.Error(fmt.Sprintf(`There was an error executing query: %s`, err))
+		return "", err
+	}
+
+	// Set next token from response
+	nextToken := ""
+	if response.NextToken != nil {
+		nextToken = *response.NextToken
+	}
+
+	// Handle the case when we have more items than the limit
+	if len(response.Items) > limit {
+		// Remove the extra item
+		response.Items = response.Items[:limit]
+	}
+
+	// Unmarshal the response into the destination
+	err = attributevalue.UnmarshalListOfMapsWithOptions(response.Items, dest, func(eo *attributevalue.DecoderOptions) {
+		eo.TagKey = "json"
+	})
+	if err != nil {
+		return "", fmt.Errorf("failed to unmarshal response: %w", err)
+	}
+
+	slog.Debug(fmt.Sprintf(`Next token: %s`, nextToken))
+	return nextToken, nil
+}
 
 func (s *DynamoDBAdapter) getTableName(obj any) string {
 	// Get the type of obj

storage/memory.go

@@ -68,3 +68,8 @@ func (m *MemoryAdapter) Delete(item any, filter map[string]any) error {
 func (m *MemoryAdapter) List(dest any, sortKey string, filter map[string]any, limit int, cursor string) (string, error) {
 	return m.DB.List(dest, sortKey, filter, limit, cursor)
 }
+
+func (m *MemoryAdapter) Search(dest any, sortKey string, query string, limit int, cursor string) (string, error) {
+	// TODO: Finish implementation
+	return "", nil
+}