Go: XGROUP CREATE.

Signed-off-by: Yury-Fridlyand <yury.fridlyand@improving.com>

Author: Yury-Fridlyand

go/api/base_client.go

@@ -1995,3 +1995,64 @@ func (client *baseClient) XPendingWithOptions(
 	}
 	return handleXPendingDetailResponse(result)
 }
+
+// Creates a new consumer group uniquely identified by `groupname` for the stream stored at `key`.
+//
+// See [valkey.io] for details.
+//
+// Parameters:
+//
+//	key - The key of the stream.
+//	group - The newly created consumer group name.
+//	id - Stream entry ID that specifies the last delivered entry in the stream from the new
+//	    group’s perspective. The special ID `"$"` can be used to specify the last entry in the stream.
+//
+// Return value:
+//
+//	`"OK"`.
+//
+// Example:
+//
+//	client.XGroupCreate("mystream", "mygroup", "0-0")
+//
+// [valkey.io]: https://valkey.io/commands/xgroup-create/
+func (client *baseClient) XGroupCreate(key string, group string, id string) (string, error) {
+	return client.XGroupCreateWithOptions(key, group, id, options.NewXGroupCreateOptions())
+}
+
+// Creates a new consumer group uniquely identified by `groupname` for the stream stored at `key`.
+//
+// See [valkey.io] for details.
+//
+// Parameters:
+//
+//	key - The key of the stream.
+//	group - The newly created consumer group name.
+//	id - Stream entry ID that specifies the last delivered entry in the stream from the new
+//	    group's perspective. The special ID `"$"` can be used to specify the last entry in the stream.
+//	opts - The options for the command. See [options.XGroupCreateOptions] for details.
+//
+// Return value:
+//
+//	`"OK"`.
+//
+// Example:
+//
+//	opts := options.NewXGroupCreateOptions().SetMakeStream()
+//	client.XGroupCreateWithOptions("mystream", "mygroup", "0-0", opts)
+//
+// [valkey.io]: https://valkey.io/commands/xgroup-create/
+func (client *baseClient) XGroupCreateWithOptions(
+	key string,
+	group string,
+	id string,
+	opts *options.XGroupCreateOptions,
+) (string, error) {
+	optionArgs, _ := opts.ToArgs()
+	args := append([]string{key, group, id}, optionArgs...)
+	result, err := client.executeCommand(C.XGroupCreate, args)
+	if err != nil {
+		return defaultStringResponse, err
+	}
+	return handleStringResponse(result)
+}

go/api/options/stream_options.go

@@ -203,3 +203,43 @@ func (xpo *XPendingOptions) ToArgs() ([]string, error) {
 
 	return args, nil
 }
+
+// Optional arguments for `XGroupCreate` in [StreamCommands]
+type XGroupCreateOptions struct {
+	mkStream    bool
+	entriesRead int64
+}
+
+// Create new empty `XGroupCreateOptions`
+func NewXGroupCreateOptions() *XGroupCreateOptions {
+	return &XGroupCreateOptions{false, -1}
+}
+
+// Once set and if the stream doesn't exist, creates a new stream with a length of `0`.
+func (xgco *XGroupCreateOptions) SetMakeStream() *XGroupCreateOptions {
+	xgco.mkStream = true
+	return xgco
+}
+
+// A value representing the number of stream entries already read by the group.
+//
+// Since Valkey version 7.0.0.
+func (xgco *XGroupCreateOptions) SetEntriesRead(entriesRead int64) *XGroupCreateOptions {
+	xgco.entriesRead = entriesRead
+	return xgco
+}
+
+func (xgco *XGroupCreateOptions) ToArgs() ([]string, error) {
+	var args []string
+
+	// if minIdleTime is set, we need to add an `IDLE` argument along with the minIdleTime
+	if xgco.mkStream {
+		args = append(args, "MKSTREAM")
+	}
+
+	if xgco.entriesRead > -1 {
+		args = append(args, "ENTRIESREAD", utils.IntToString(xgco.entriesRead))
+	}
+
+	return args, nil
+}

go/api/stream_commands.go

@@ -111,4 +111,8 @@ type StreamCommands interface {
 	XPending(key string, group string) (XPendingSummary, error)
 
 	XPendingWithOptions(key string, group string, options *options.XPendingOptions) ([]XPendingDetail, error)
+
+	XGroupCreate(key string, group string, id string) (string, error)
+
+	XGroupCreateWithOptions(key string, group string, id string, opts *options.XGroupCreateOptions) (string, error)
 }

go/integTest/shared_commands_test.go

@@ -5716,3 +5716,46 @@ func (suite *GlideTestSuite) TestXPendingFailures() {
 		}
 	})
 }
+
+// TODO add XGroupDestroy tests there
+func (suite *GlideTestSuite) TestXGroupCreate_XGroupDestroy() {
+	suite.runWithDefaultClients(func(client api.BaseClient) {
+		key := uuid.NewString()
+		group1 := uuid.NewString()
+		group2 := uuid.NewString()
+		id := "0-1"
+
+		// Stream not created results in error
+		_, err := client.XGroupCreate(key, group1, id)
+		assert.Error(suite.T(), err)
+		assert.IsType(suite.T(), &api.RequestError{}, err)
+
+		// Stream with option to create creates stream & Group
+		opts := options.NewXGroupCreateOptions().SetMakeStream()
+		suite.verifyOK(client.XGroupCreateWithOptions(key, group1, id, opts))
+
+		// ...and again results in BUSYGROUP error, because group names must be unique
+		_, err = client.XGroupCreate(key, group1, id)
+		assert.ErrorContains(suite.T(), err, "BUSYGROUP")
+		assert.IsType(suite.T(), &api.RequestError{}, err)
+
+		// TODO add XGroupDestroy tests there
+
+		// ENTRIESREAD option was added in valkey 7.0.0
+		opts = options.NewXGroupCreateOptions().SetEntriesRead(100)
+		if suite.serverVersion >= "7.0.0" {
+			suite.verifyOK(client.XGroupCreateWithOptions(key, group2, id, opts))
+		} else {
+			_, err = client.XGroupCreateWithOptions(key, group2, id, opts)
+			assert.Error(suite.T(), err)
+			assert.IsType(suite.T(), &api.RequestError{}, err)
+		}
+
+		// key is not a stream
+		key = uuid.NewString()
+		suite.verifyOK(client.Set(key, id))
+		_, err = client.XGroupCreate(key, group1, id)
+		assert.Error(suite.T(), err)
+		assert.IsType(suite.T(), &api.RequestError{}, err)
+	})
+}