Assorted fixes

query/cursor.go

@@ -1,6 +1,7 @@
 package query
 
 import (
+	"encoding/base64"
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
@@ -100,9 +101,22 @@ func (opts *QueryOptions) DecodeCursor(cursorStr string) (*Cursor, error) {
 			return nil, fmt.Errorf("unknown value for descending: %d", descendingInt)
 		}
 
+		value := c[1]
+		if opts.Columns[columnIx].IsBytes {
+			encoded, ok := value.(string)
+			if !ok {
+				return nil, fmt.Errorf("failed to decode cursor, byte column is not a string")
+			}
+			decoded, err := base64.StdEncoding.DecodeString(encoded)
+			if err != nil {
+				return nil, fmt.Errorf("failed to decode cursor: %w", err)
+			}
+			value = decoded
+		}
+
 		cursor.ColumnValues = append(cursor.ColumnValues, CursorValue{
 			ColumnName: opts.Columns[columnIx].Name,
-			Value:      c[1],
+			Value:      value,
 			Descending: descending,
 		})
 	}

query/query.go

@@ -52,6 +52,12 @@ func (b *QueryBuilder) addPaginationArguments() error {
 	paginationConditions := []string{}
 
 	if len(b.options.Cursor) > 0 {
+		// Pre-allocate argument counters so that we don't need to duplicate them below
+		args := make([]string, 0, len(b.options.Cursor))
+		for _, val := range b.options.Cursor {
+			args = append(args, b.pushArgument(val.Value))
+		}
+
 		for i := range b.options.Cursor {
 			subcondition := []string{}
 			for j, from := range b.options.Cursor {
@@ -67,7 +73,7 @@ func (b *QueryBuilder) addPaginationArguments() error {
 					operator = ">"
 				}
 
-				arg := b.pushArgument(from.Value)
+				arg := args[j]
 
 				columnIx, err := b.options.GetColumnIndex(from.ColumnName)
 				if err != nil {
@@ -96,7 +102,9 @@ func (b *QueryBuilder) addPaginationArguments() error {
 			b.queryBuilder.WriteString(" AND ")
 		}
 
+		b.queryBuilder.WriteString("(")
 		b.queryBuilder.WriteString(paginationCondition)
+		b.queryBuilder.WriteString(")")
 	}
 
 	return nil

query/query_options.go

@@ -10,16 +10,18 @@ import (
 
 const QueryResultCountLimit uint64 = 200
 
-// 256 bytes is for the overhead of the 'envelope' around the entity data
+// ResponseSize is 256 bytes for the overhead of the 'envelope' around the entity data
 // and the separator characters in between
 const ResponseSize int = 256
 
-// 512 kb
+// MaxResponseSize is 512 kb
 const MaxResponseSize int = 512 * 1024 * 1024
 
 type Column struct {
 	Name          string
 	QualifiedName string
+	// If this is a byte column, we need to decode it when we get it from the json-encoded cursor
+	IsBytes bool
 }
 
 func (c Column) selector() string {
@@ -67,6 +69,7 @@ func NewQueryOptions(log *slog.Logger, latestHead uint64, options *InternalQuery
 	queryOptions.Columns = append(queryOptions.Columns, Column{
 		Name:          "entity_key",
 		QualifiedName: "e.entity_key",
+		IsBytes:       true,
 	})
 
 	if options.IncludeData.Payload {
@@ -151,6 +154,7 @@ func NewQueryOptions(log *slog.Logger, latestHead uint64, options *InternalQuery
 		Column: Column{
 			Name:          "entity_key",
 			QualifiedName: "e.entity_key",
+			IsBytes:       true,
 		},
 	})
 

query/types.go

@@ -7,20 +7,6 @@ import (
 	"github.com/ethereum/go-ethereum/common/hexutil"
 )
 
-var defaultColumns map[string]string
-
-func init() {
-	columns := []string{
-		"entity_key",
-		"from_block",
-	}
-
-	defaultColumns = make(map[string]string, len(columns))
-	for _, column := range columns {
-		defaultColumns[column] = column
-	}
-}
-
 var KeyAttributeKey = "$key"
 var CreatorAttributeKey = "$creator"
 var OwnerAttributeKey = "$owner"

sqlstore/sqlstore.go

@@ -195,10 +195,7 @@ func (s *SQLStore) QueryEntitiesInternalIterator(
 ) error {
 	s.log.Info("Executing query", "query", queryStr, "args", args)
 
-	queryCtx, queryCtxCancel := context.WithCancel(ctx)
-	defer queryCtxCancel()
-
-	rows, err := s.db.QueryContext(queryCtx, queryStr, args...)
+	rows, err := s.db.QueryContext(ctx, queryStr, args...)
 	if err != nil {
 		return fmt.Errorf("failed to get entities for query: %s: %w", queryStr, err)
 	}