Go: XGROUP DESTROY.

go/api/base_client.go

@@ -1995,3 +1995,90 @@ func (client *baseClient) XPendingWithOptions(
 	}
 	return handleXPendingDetailResponse(result)
 }
+
+// Creates a new consumer group uniquely identified by `group` 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 `group` 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)
+}
+
+// Destroys the consumer group `group` for the stream stored at `key`.
+//
+// See [valkey.io] for details.
+//
+// Parameters:
+//
+//	key - The key of the stream.
+//	group - The consumer group name to delete.
+//
+// Return value:
+//
+//	`true` if the consumer group is destroyed. Otherwise, `false`.
+//
+// Example:
+//
+//	client.XGroupDestroy("mystream", "mygroup")
+//
+// [valkey.io]: https://valkey.io/commands/xgroup-destroy/
+func (client *baseClient) XGroupDestroy(key string, group string) (bool, error) {
+	result, err := client.executeCommand(C.XGroupDestroy, []string{key, group})
+	if err != nil {
+		return defaultBoolResponse, err
+	}
+	return handleBoolResponse(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,10 @@ 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)
+
+	XGroupDestroy(key string, group string) (bool, error)
 }

go/integTest/shared_commands_test.go

@@ -5407,15 +5407,13 @@ func (suite *GlideTestSuite) TestXPendingFailures() {
 			consumer1 := "consumer-1-" + uuid.New().String()
 			invalidConsumer := "invalid-consumer-" + uuid.New().String()
 
-			command := []string{"XGroup", "Create", key, groupName, zeroStreamId, "MKSTREAM"}
+			suite.verifyOK(
+				client.XGroupCreateWithOptions(key, groupName, zeroStreamId, options.NewXGroupCreateOptions().SetMakeStream()),
+			)
 
+			command := []string{"XGroup", "CreateConsumer", key, groupName, consumer1}
 			resp, err := client.CustomCommand(command)
 			assert.NoError(suite.T(), err)
-			assert.Equal(suite.T(), "OK", resp.(string))
-
-			command = []string{"XGroup", "CreateConsumer", key, groupName, consumer1}
-			resp, err = client.CustomCommand(command)
-			assert.NoError(suite.T(), err)
 			assert.True(suite.T(), resp.(bool))
 
 			_, err = client.XAdd(key, [][]string{{"field1", "value1"}})
@@ -5560,15 +5558,13 @@ func (suite *GlideTestSuite) TestXPendingFailures() {
 			consumer1 := "consumer-1-" + uuid.New().String()
 			invalidConsumer := "invalid-consumer-" + uuid.New().String()
 
-			command := []string{"XGroup", "Create", key, groupName, zeroStreamId, "MKSTREAM"}
+			suite.verifyOK(
+				client.XGroupCreateWithOptions(key, groupName, zeroStreamId, options.NewXGroupCreateOptions().SetMakeStream()),
+			)
 
+			command := []string{"XGroup", "CreateConsumer", key, groupName, consumer1}
 			resp, err := client.CustomCommand(command)
 			assert.NoError(suite.T(), err)
-			assert.Equal(suite.T(), "OK", resp.Value().(string))
-
-			command = []string{"XGroup", "CreateConsumer", key, groupName, consumer1}
-			resp, err = client.CustomCommand(command)
-			assert.NoError(suite.T(), err)
 			assert.True(suite.T(), resp.Value().(bool))
 
 			_, err = client.XAdd(key, [][]string{{"field1", "value1"}})
@@ -5716,3 +5712,52 @@ func (suite *GlideTestSuite) TestXPendingFailures() {
 		}
 	})
 }
+
+func (suite *GlideTestSuite) TestXGroupCreate_XGroupDestroy() {
+	suite.runWithDefaultClients(func(client api.BaseClient) {
+		key := uuid.NewString()
+		group := uuid.NewString()
+		id := "0-1"
+
+		// Stream not created results in error
+		_, err := client.XGroupCreate(key, group, 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, group, id, opts))
+
+		// ...and again results in BUSYGROUP error, because group names must be unique
+		_, err = client.XGroupCreate(key, group, id)
+		assert.ErrorContains(suite.T(), err, "BUSYGROUP")
+		assert.IsType(suite.T(), &api.RequestError{}, err)
+
+		// Stream Group can be destroyed returns: true
+		destroyed, err := client.XGroupDestroy(key, group)
+		assert.NoError(suite.T(), err)
+		assert.True(suite.T(), destroyed)
+
+		// ...and again results in: false
+		destroyed, err = client.XGroupDestroy(key, group)
+		assert.NoError(suite.T(), err)
+		assert.False(suite.T(), destroyed)
+
+		// 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, group, id, opts))
+		} else {
+			_, err = client.XGroupCreateWithOptions(key, group, 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, group, id)
+		assert.Error(suite.T(), err)
+		assert.IsType(suite.T(), &api.RequestError{}, err)
+	})
+}