ronantakizawa/codereview-bench huggingface dataset:
Repository: DiceDB/dice
File: server/async_tcp.go (Go)

Code before:
// FindPortAndBind binds the server to the given host and port
func (s *AsyncServer) FindPortAndBind() error {
	serverFD, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_STREAM, 0)
	if err != nil {
		return err
	}
	s.serverFD = serverFD

	if err := syscall.SetsockoptInt(serverFD, syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1); err != nil {
		return err
	}

	if err := syscall.SetNonblock(serverFD, true); err != nil {
		return err
	}

	ip4 := net.ParseIP(config.Host)

	return syscall.Bind(serverFD, &syscall.SockaddrInet4{
		Port: config.Port,
		Addr: [4]byte{ip4[0], ip4[1], ip4[2], ip4[3]},
	})
}

// Run starts the server, accepts connections, and handles client requests
func (s *AsyncServer) Run() error {
	defer s.wg.Done()
	defer syscall.Close(s.serverFD)

	if err := s.SetupUsers(); err != nil {
		return err
	}

	s.wg.Add(1)
	go s.WatchKeys()

	if err := syscall.Listen(s.serverFD, s.maxClients); err != nil {
		return err
	}

	var err error
	s.multiplexer, err = iomultiplexer.New(s.maxClients)
	if err != nil {
		return err
	}
	defer s.multiplexer.Close()

	if err := s.multiplexer.Subscribe(iomultiplexer.Event{
		Fd: s.serverFD,
		Op: iomultiplexer.OpRead,
	}); err != nil {

Review comment (suggestion):
Same thought, we can pass the context here. 
```suggestion
func (s *AsyncServer) Run(ctx context.Context) error {
```

Code after:
	if err != nil {
		return err
	}
	s.serverFD = serverFD

	// Close the socket on exit if an error occurs
	defer func() {
		if err != nil {
			if err := syscall.Close(serverFD); err != nil {
				log.Warn("failed to close server socket", "error", err)
			}
		}
	}()

	if err := syscall.SetsockoptInt(serverFD, syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1); err != nil {
		return err
	}

	if err := syscall.SetNonblock(serverFD, true); err != nil {
		return err
	}

	ip4 := net.ParseIP(config.Host)
	if ip4 == nil {
		return ErrInvalidIPAddress
	}

	return syscall.Bind(serverFD, &syscall.SockaddrInet4{
		Port: config.Port,
		Addr: [4]byte{ip4[0], ip4[1], ip4[2], ip4[3]},
	})
}

// Run starts the server, accepts connections, and handles client requests
func (s *AsyncServer) Run(ctx context.Context) error {
	defer func(fd int) {
		err := syscall.Close(fd)
		if err != nil {
			log.Warn("failed to close server socket", "error", err)
		}
	}(s.serverFD)

	watchCtx, cancelWatch := context.WithCancel(ctx)
	defer cancelWatch()

	if err := s.SetupUsers(); err != nil {
		return err
	}

	var wg sync.WaitGroup
	wg.Add(1)
Repository: DiceDB/dice
File: core/eval.go (Go)

Code before:
	}

	store.PutAll(insertMap)
	return RespOK
}

// evalGET returns the value for the queried key in args
// The key should be the only param in args
// The RESP value of the key is encoded and then returned
// evalGET returns RespNIL if key is expired or it does not exist
func evalGET(args []string, store *Store) []byte {
	if len(args) != 1 {
		return diceerrors.NewErrArity("GET")
	}

	var key = args[0]

	obj := store.Get(key)

	// if key does not exist, return RESP encoded nil
	if obj == nil {
		return RespNIL
	}

	// If the object exists, check if it is a set object.
	if err := assertType(obj.TypeEncoding, ObjTypeSet); err == nil {
		return diceerrors.NewErrSetType()
	}

	// Decode and return the value based on its encoding
	switch _, oEnc := ExtractTypeEncoding(obj); oEnc {
	case ObjEncodingInt:
		// Value is stored as an int64, so use type assertion
		if val, ok := obj.Value.(int64); ok {
			return Encode(val, false)
		}
		return diceerrors.NewErrWithFormattedMessage("expected int64 but got another type: %s", obj.Value)

	case ObjEncodingEmbStr, ObjEncodingRaw:
		// Value is stored as a string, use type assertion
		if val, ok := obj.Value.(string); ok {
			return Encode(val, false)
		}
		return diceerrors.NewErrWithMessage("expected string but got another type")

	default:
		return diceerrors.NewErrWithMessage(diceerrors.WrongTypeErr)
	}
}

// evalDBSIZE returns the number of keys in the database.

Review comment (refactor):
Shall L270 i.e `switch _, oEnc := ExtractTypeEncoding(obj); oEnc` shouldn't handle this case?
Any specific reasons to introduce this check here?

Code after:
	}

	store.PutAll(insertMap)
	return RespOK
}

// evalGET returns the value for the queried key in args
// The key should be the only param in args
// The RESP value of the key is encoded and then returned
// evalGET returns RespNIL if key is expired or it does not exist
func evalGET(args []string, store *Store) []byte {
	if len(args) != 1 {
		return diceerrors.NewErrArity("GET")
	}

	var key = args[0]

	obj := store.Get(key)

	// if key does not exist, return RESP encoded nil
	if obj == nil {
		return RespNIL
	}

	// Decode and return the value based on its encoding
	switch _, oEnc := ExtractTypeEncoding(obj); oEnc {
	case ObjEncodingInt:
		// Value is stored as an int64, so use type assertion
		if val, ok := obj.Value.(int64); ok {
			return Encode(val, false)
		}
		return diceerrors.NewErrWithFormattedMessage("expected int64 but got another type: %s", obj.Value)

	case ObjEncodingEmbStr, ObjEncodingRaw:
		// Value is stored as a string, use type assertion
		if val, ok := obj.Value.(string); ok {
			return Encode(val, false)
		}
		return diceerrors.NewErrWithMessage("expected string but got another type")

	default:
		return diceerrors.NewErrWithMessage(diceerrors.WrongTypeErr)
	}
}

// evalDBSIZE returns the number of keys in the database.
func evalDBSIZE(args []string, store *Store) []byte {
	if len(args) > 0 {
		return diceerrors.NewErrArity("DBSIZE")
	}
Repository: DiceDB/dice
File: internal/clientio/resp.go (Go)

Code before:
		buf := bytes.NewBuffer(b)
		for _, b := range value.([]int64) {
			buf.Write(Encode(b, false))
		}
		return []byte(fmt.Sprintf("*%d\r\n%s", len(v), buf.Bytes()))
	case error:
		return []byte(fmt.Sprintf("-%s\r\n", v))
	case dstore.WatchEvent:
		var b []byte
		buf := bytes.NewBuffer(b)
		we := value.(dstore.WatchEvent)
		buf.Write(Encode(fmt.Sprintf("key:%s", we.Key), false))
		buf.Write(Encode(fmt.Sprintf("op:%s", we.Operation), false))
		return []byte(fmt.Sprintf("*2\r\n%s", buf.Bytes()))
	case []dstore.DSQLQueryResultRow:
		var b []byte
		buf := bytes.NewBuffer(b)
		for _, row := range value.([]dstore.DSQLQueryResultRow) {
			buf.WriteString("*2\r\n")
			buf.Write(Encode(row.Key, false))
			buf.Write(Encode(row.Value.Value, false))
		}
		return []byte(fmt.Sprintf("*%d\r\n%s", len(v), buf.Bytes()))
	case map[string]bool:
		return RespNIL
	case []DiceCmdMeta:
		var b []byte
		buf := bytes.NewBuffer(b)
		for _, b := range value.([]DiceCmdMeta) {
			buf.WriteString("*3\r\n")
			buf.Write(encodeString(b.Name))
			buf.Write(Encode(b.Arity, isSimple))
			buf.Write(Encode(b.KeySpecs, isSimple))
		}
		return []byte(fmt.Sprintf("*%d\r\n%s", len(v), buf.Bytes()))
	case KeySpecs:
		var b []byte
		buf := bytes.NewBuffer(b)
		buf.Write(encodeString("BeginIndex"))
		buf.Write(Encode(v.BeginIndex, isSimple))
		buf.Write(encodeString("LastKey"))
		buf.Write(Encode(v.LastKey, isSimple))
		buf.Write(encodeString("Step"))
		buf.Write(Encode(v.Step, isSimple))
		return []byte(fmt.Sprintf("*6\r\n%s", buf.Bytes()))
	default:
		fmt.Printf("Unsupported type: %T\n", v)
		return RespNIL
	}
}


Review comment (suggestion):
@aashraybhandar1 can you please rebase once and validate, this call would error out as `DiceCmdMeta` is part of `eval` now.
Also I believe we should keep type as `[]interface{}` we have similar response in `QWATCH` you can take a reference of `CreatePushResponse` function may be.

Code after:
		buf := bytes.NewBuffer(b)
		for _, b := range value.([]int64) {
			buf.Write(Encode(b, false))
		}
		return []byte(fmt.Sprintf("*%d\r\n%s", len(v), buf.Bytes()))
	case error:
		return []byte(fmt.Sprintf("-%s\r\n", v))
	case dstore.WatchEvent:
		var b []byte
		buf := bytes.NewBuffer(b)
		we := value.(dstore.WatchEvent)
		buf.Write(Encode(fmt.Sprintf("key:%s", we.Key), false))
		buf.Write(Encode(fmt.Sprintf("op:%s", we.Operation), false))
		return []byte(fmt.Sprintf("*2\r\n%s", buf.Bytes()))
	case []dstore.DSQLQueryResultRow:
		var b []byte
		buf := bytes.NewBuffer(b)
		for _, row := range value.([]dstore.DSQLQueryResultRow) {
			buf.WriteString("*2\r\n")
			buf.Write(Encode(row.Key, false))
			buf.Write(Encode(row.Value.Value, false))
		}
		return []byte(fmt.Sprintf("*%d\r\n%s", len(v), buf.Bytes()))
	case map[string]bool:
		return RespNIL
	default:
		fmt.Printf("Unsupported type: %T\n", v)
		return RespNIL
	}
}
Repository: DiceDB/dice
File: internal/querywatcher/query_watcher.go (Go)

Code before:
			} else {
				w.removeWatcher(&event.Query, event.ClientFD)
			}
		case <-ctx.Done():
			return
		}
	}
}

// watchKeys watches for changes in keys and notifies clients.
func (w *QWatchManager) watchKeys(ctx context.Context, watchChan <-chan dstore.WatchEvent) {
	for {
		select {
		case event := <-watchChan:
			w.processWatchEvent(event)
		case <-ctx.Done():
			return
		}
	}
}

// processWatchEvent processes a single watch event.
func (w *QWatchManager) processWatchEvent(event dstore.WatchEvent) {
	/*
		Iterate over the watchlist to go through all fingerprints and the corresponding client connections to those fingerprints
	*/
	w.WatchList.Range(func(fingerPrint, value interface{}) bool {
		/*
			Fetch all the queries that match the fingerprint. If the fingerprint is not found in the cache, then it is likely that the query has been deleted. In that case, we can skip the rest of the logic
		*/
		queries, ok := w.FingerPrintCache.Get(fingerPrint.(string))
		if !ok {
			log.Warnf("Fingerprint not found in cacheStore: %s", fingerPrint)
			return true
		}

		clients := value.(*sync.Map)
		/*
			For all the clients that are watching this fingerprint, we need to find the queries they had requested to run, and then send them responses based on the queries they had fired.
		*/
		clients.Range(func(clientKey, _ interface{}) bool {
			clientFd := clientKey.(int)

			query := queries[clientFd]
			if !regex.WildCardMatch(query.KeyRegex, event.Key) {
				return true
			}

			w.updateQueryCache(fingerPrint.(string), event)
			queryResult, err := w.runQuery(query)
			if err != nil {

Review comment (suggestion):
nit: shouldn't we use `//` style comments here? I am not familiar with what the recommend styling guide is for comments in golang, so I'm happy to be corrected here.

Code after:
		}
	}
}

// watchKeys watches for changes in keys and notifies clients.
func (w *QueryManager) watchKeys(ctx context.Context, watchChan <-chan dstore.WatchEvent) {
	for {
		select {
		case event := <-watchChan:
			w.processWatchEvent(event)
		case <-ctx.Done():
			return
		}
	}
}

// processWatchEvent processes a single watch event.
func (w *QueryManager) processWatchEvent(event dstore.WatchEvent) {
	// Iterate over the watchlist to go through the query string
	// and the corresponding client connections to that query string
	w.WatchList.Range(func(key, value interface{}) bool {
		queryString := key.(string)
		clients := value.(*sync.Map)

		query, err := ParseQuery(queryString)
		if err != nil {
			log.Error(fmt.Sprintf("error parsing query: %s", queryString))
			return true
		}

		// Check if the key matches the regex
		if !regex.WildCardMatch(query.KeyRegex, event.Key) {
			return true
		}

		w.updateQueryCache(query.Fingerprint, event)

		queryResult, err := w.runQuery(&query)
		if err != nil {
			log.Error(err)
			return true
		}

		w.notifyClients(&query, clients, queryResult)
		return true
	})
}

// updateQueryCache updates the query cache based on the watch event.
func (w *QueryManager) updateQueryCache(queryFingerprint string, event dstore.WatchEvent) {
	w.QueryCacheMu.Lock()
Repository: DiceDB/dice
File: internal/querywatcher/query_watcher.go (Go)

Code before:
			} else {
				w.removeWatcher(&event.Query, event.ClientFD)
			}
		case <-ctx.Done():
			return
		}
	}
}

// watchKeys watches for changes in keys and notifies clients.
func (w *QueryManager) watchKeys(ctx context.Context, watchChan <-chan dstore.WatchEvent) {
	for {
		select {
		case event := <-watchChan:
			w.processWatchEvent(event)
		case <-ctx.Done():
			return
		}
	}
}

// processWatchEvent processes a single watch event.
func (w *QueryManager) processWatchEvent(event dstore.WatchEvent) {
	// Iterate over the watchlist to go through the query string
	// and the corresponding client connections to that query string
	w.WatchList.Range(func(key, value interface{}) bool {
		queryString := key.(string)
		clients := value.(*sync.Map)

		query, err := ParseQuery(queryString)
		if err != nil {
			log.Error(fmt.Sprintf("Error parsing query: %s", queryString))
			return true
		}

		// Check if the key matches the regex
		if !regex.WildCardMatch(query.KeyRegex, event.Key) {
			return true
		}

		w.updateQueryCache(query.Fingerprint, event)

		queryResult, err := w.runQuery(&query)
		if err != nil {
			log.Error(err)
			return true
		}

		w.notifyClients(&query, clients, queryResult)
		return true
	})

Review comment (suggestion):
For my understanding, why are we changing the key of the WatchList to a string? Isn't it possible to keep it as a `DSQLQuery` object? That would save us a ParseQuery call on every execution.

Code after:
				w.removeWatcher(&event.Query, event.ClientFD)
			}
		case <-ctx.Done():
			return
		}
	}
}

// watchKeys watches for changes in keys and notifies clients.
func (w *QueryManager) watchKeys(ctx context.Context, watchChan <-chan dstore.WatchEvent) {
	for {
		select {
		case event := <-watchChan:
			w.processWatchEvent(event)
		case <-ctx.Done():
			return
		}
	}
}

// processWatchEvent processes a single watch event.
func (w *QueryManager) processWatchEvent(event dstore.WatchEvent) {
	// Iterate over the watchlist to go through the query string
	// and the corresponding client connections to that query string
	w.WatchList.Range(func(key, value interface{}) bool {
		queryString := key.(string)
		clients := value.(*sync.Map)

		query, err := ParseQuery(queryString)
		if err != nil {
			log.Error(fmt.Sprintf("error parsing query: %s", queryString))
			return true
		}

		// Check if the key matches the regex
		if !regex.WildCardMatch(query.KeyRegex, event.Key) {
			return true
		}

		w.updateQueryCache(query.Fingerprint, event)

		queryResult, err := w.runQuery(&query)
		if err != nil {
			log.Error(err)
			return true
		}

		w.notifyClients(&query, clients, queryResult)
		return true
	})
}
Repository: DiceDB/dice
File: internal/shard/shard_thread.go (Go)

Code before:
		return diceerrors.NewErrWithFormattedMessage("unknown command '%s', with args beginning with: %s", op.Cmd.Cmd, strings.Join(op.Cmd.Args, " "))
	}

	// The following commands could be handled at the server level, however, we can randomly let any shard handle them
	// to reduce load on main server.
	switch diceCmd.Name {
	case "SUBSCRIBE", "QWATCH":
		return eval.EvalQWATCH(op.Cmd.Args, op.Client.Fd, shard.store)
	case "UNSUBSCRIBE", "QUNWATCH":
		return eval.EvalQUNWATCH(op.Cmd.Args, op.Client.Fd)
	case auth.AuthCmd:
		return eval.EvalAUTH(op.Cmd.Args, op.Client)
	case "ABORT":
		return clientio.RespOK
	default:
		return diceCmd.Eval(op.Cmd.Args, shard.store)
	}
}

// cleanup handles cleanup logic when the shard stops.
func (shard *ShardThread) cleanup() {
	close(shard.ReqChan)
	if testing.Testing() {
		// Avoiding AOF dump in case of testing env
		return
	}

	eval.EvalBGREWRITEAOF([]string{}, shard.store)
}


Review comment (suggestion):
Thanks for tackling this issue as a part of this PR. It would be better if we make this configurable using a config variable instead of adding a static condition in this manner.

Additionally, when you set that config for the test server inside the setup method, please add an explanation as to why we're doing it.

Code after:
	if !ok {
		return diceerrors.NewErrWithFormattedMessage("unknown command '%s', with args beginning with: %s", op.Cmd.Cmd, strings.Join(op.Cmd.Args, " "))
	}

	// The following commands could be handled at the server level, however, we can randomly let any shard handle them
	// to reduce load on main server.
	switch diceCmd.Name {
	case "SUBSCRIBE", "QWATCH":
		return eval.EvalQWATCH(op.Cmd.Args, op.Client.Fd, shard.store)
	case "UNSUBSCRIBE", "QUNWATCH":
		return eval.EvalQUNWATCH(op.Cmd.Args, op.Client.Fd)
	case auth.AuthCmd:
		return eval.EvalAUTH(op.Cmd.Args, op.Client)
	case "ABORT":
		return clientio.RespOK
	default:
		return diceCmd.Eval(op.Cmd.Args, shard.store)
	}
}

// cleanup handles cleanup logic when the shard stops.
func (shard *ShardThread) cleanup() {
	close(shard.ReqChan)
	if config.WriteAOFOnCleanup {
		// Avoiding AOF dump for test enabled environments as
		// the tests were taking longer due to background tasks which exceeded the WaitDelay,
		// thus causing the test process to be forcibly terminated.
		log.Info("Skipping AOF dump as test env enabled.")
		return
	}

	eval.EvalBGREWRITEAOF([]string{}, shard.store)
}
Repository: DiceDB/dice
File: internal/server/utils/redisCmdAdapter.go (Go)

Code before:
package utils

import (
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"

	"github.com/dicedb/dice/internal/cmd"
)

func ParseHTTPRequest(r *http.Request) (*cmd.RedisCmd, error) {
	prefix := strings.TrimPrefix(r.URL.Path, "/")
	if prefix == "" {
		return nil, errors.New("invalid command")
	}

	prefix = strings.ToUpper(prefix)
	var args []string

	// Step 1: Handle JSON body if present
	if r.Body != nil {
		body, err := io.ReadAll(r.Body)
		if err != nil {
			return nil, err
		}

		if len(body) > 0 {
			var jsonBody map[string]interface{}
			if err := json.Unmarshal(body, &jsonBody); err != nil {
				return nil, err
			}

			// Define keys to exclude and process their values first
			// Update as we support more commands
			priorityKeys := []string{"key", "field", "path", "value"}
			for _, key := range priorityKeys {
				if val, exists := jsonBody[key]; exists {

Review comment (suggestion):
nit: just for better clarity.
```suggestion
	command := strings.TrimPrefix(r.URL.Path, "/")
```

Code after:
package utils

import (
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"

	"github.com/dicedb/dice/internal/cmd"
)

const (
	Key   = "key"
	Field = "field"
	Path  = "path"
	Value = "value"
)

func ParseHTTPRequest(r *http.Request) (*cmd.RedisCmd, error) {
	command := strings.TrimPrefix(r.URL.Path, "/")
	if command == "" {
		return nil, errors.New("invalid command")
	}

	command = strings.ToUpper(command)
	var args []string

	// Step 1: Handle JSON body if present
	if r.Body != nil {
		body, err := io.ReadAll(r.Body)
		if err != nil {
			return nil, err
		}

		if len(body) > 0 {
			var jsonBody map[string]interface{}
			if err := json.Unmarshal(body, &jsonBody); err != nil {
				return nil, err
Repository: DiceDB/dice
File: internal/eval/eval.go (Go)

Code before:
	values := args[2:]

	obj := store.Get(key)
	if obj == nil {
		return diceerrors.NewErrWithMessage("ERR key does not exist")
	}

	errWithMessage := object.AssertTypeAndEncoding(obj.TypeEncoding, object.ObjTypeJSON, object.ObjEncodingJSON)
	if errWithMessage != nil {
		return errWithMessage
	}

	jsonData := obj.Value

	expr, err := jp.ParseString(path)
	if err != nil {
		return diceerrors.NewErrWithMessage("ERR Path '<path>' does not exist or not an array")
	}

	// Parse the input values as JSON
	parsedValues := make([]interface{}, len(values))
	for i, v := range values {
		var parsedValue interface{}
		err := sonic.UnmarshalString(v, &parsedValue)
		if err != nil {
			return diceerrors.NewErrWithMessage(fmt.Sprintf("ERR invalid JSON value: %s", v))
		}
		parsedValues[i] = parsedValue
	}

	var resultsArray []interface{}
	modified := false

	// Capture the modified data when modifying the root path
	var newData interface{}
	var modifyErr error

	newData, modifyErr = expr.Modify(jsonData, func(data any) (interface{}, bool) {
		arr, ok := data.([]interface{})
		if !ok {
			// Not an array
			resultsArray = append(resultsArray, nil)
			return data, false
		}

		// Append the parsed values to the array
		arr = append(arr, parsedValues...)

		resultsArray = append(resultsArray, int64(len(arr)))
		modified = true
		return arr, modified

Review comment (suggestion):
```suggestion
			return diceerrors.NewErrWithMessage(err)
```

Code after:
	values := args[2:]

	obj := store.Get(key)
	if obj == nil {
		return diceerrors.NewErrWithMessage("ERR key does not exist")
	}

	errWithMessage := object.AssertTypeAndEncoding(obj.TypeEncoding, object.ObjTypeJSON, object.ObjEncodingJSON)
	if errWithMessage != nil {
		return errWithMessage
	}

	jsonData := obj.Value

	expr, err := jp.ParseString(path)
	if err != nil {
		return diceerrors.NewErrWithMessage(fmt.Sprintf("ERR Path '%s' does not exist or not an array", path))
	}

	// Parse the input values as JSON
	parsedValues := make([]interface{}, len(values))
	for i, v := range values {
		var parsedValue interface{}
		err := sonic.UnmarshalString(v, &parsedValue)
		if err != nil {
			return diceerrors.NewErrWithMessage(err.Error())
		}
		parsedValues[i] = parsedValue
	}

	var resultsArray []interface{}
	modified := false

	// Capture the modified data when modifying the root path
	var newData interface{}
	var modifyErr error

	newData, modifyErr = expr.Modify(jsonData, func(data any) (interface{}, bool) {
		arr, ok := data.([]interface{})
		if !ok {
			// Not an array
			resultsArray = append(resultsArray, nil)
			return data, false
		}

		// Append the parsed values to the array
		arr = append(arr, parsedValues...)

		resultsArray = append(resultsArray, int64(len(arr)))
		modified = true
		return arr, modified