feat: subscription create need namespace

internal/controller/trigger/controller.go

@@ -179,13 +179,6 @@ func (ctrl *controller) CreateSubscription(ctx context.Context,
 		})
 		return nil, err
 	}
-	// subscription name can't be repeated in an eventbus
-	eventbusID := vanus.NewIDFromUint64(request.Subscription.EventbusId)
-	_sub := ctrl.subscriptionManager.GetSubscriptionByName(ctx, eventbusID, request.Subscription.Name)
-	if _sub != nil {
-		return nil, errors.ErrInvalidRequest.WithMessage(
-			fmt.Sprintf("subscription name %s has exist", request.Subscription.Name))
-	}
 	sub := convert.FromPbSubscriptionRequest(request.Subscription)
 	sub.ID, err = vanus.NewID()
 	sub.CreatedAt = time.Now()
@@ -230,14 +223,6 @@ func (ctrl *controller) UpdateSubscription(ctx context.Context,
 	if request.Subscription.EventbusId != uint64(sub.EventbusID) {
 		return nil, errors.ErrInvalidRequest.WithMessage("can not change eventbus")
 	}
-	if request.Subscription.Name != sub.Name {
-		// subscription name can't be repeated in an eventbus
-		_sub := ctrl.subscriptionManager.GetSubscriptionByName(ctx, sub.EventbusID, request.Subscription.Name)
-		if _sub != nil {
-			return nil, errors.ErrInvalidRequest.WithMessage(
-				fmt.Sprintf("subscription name %s has exist", request.Subscription.Name))
-		}
-	}
 	update := convert.FromPbSubscriptionRequest(request.Subscription)
 	transChange := 0
 	if !sub.Transformer.Exist() && update.Transformer.Exist() {

internal/controller/trigger/controller_test.go

@@ -130,9 +130,10 @@ func TestController_CreateSubscription(t *testing.T) {
 			subManager.EXPECT().AddSubscription(gomock.Any(), gomock.Any()).AnyTimes().Return(nil)
 			create := &ctrlpb.CreateSubscriptionRequest{
 				Subscription: &ctrlpb.SubscriptionRequest{
-					EventbusId: vanus.NewTestID().Uint64(),
-					Name:       "test-name",
-					Sink:       "test-sink",
+					NamespaceId: vanus.NewTestID().Uint64(),
+					EventbusId:  vanus.NewTestID().Uint64(),
+					Name:        "test-name",
+					Sink:        "test-sink",
 				},
 			}
 			request := create.Subscription
@@ -141,10 +142,12 @@ func TestController_CreateSubscription(t *testing.T) {
 			So(err, ShouldBeNil)
 			So(resp.Sink, ShouldEqual, request.Sink)
 			So(resp.EventbusId, ShouldEqual, request.EventbusId)
+			So(resp.NamespaceId, ShouldEqual, request.NamespaceId)
 			resp2, err := ctrl.CreateSubscription(ctx, create)
 			So(err, ShouldBeNil)
 			So(resp2.Sink, ShouldEqual, request.Sink)
 			So(resp2.EventbusId, ShouldEqual, request.EventbusId)
+			So(resp2.NamespaceId, ShouldEqual, request.NamespaceId)
 			So(resp.Id, ShouldNotEqual, resp2.Id)
 		})
 	})
@@ -164,6 +167,7 @@ func TestController_UpdateSubscription(t *testing.T) {
 
 		subID := vanus.NewTestID()
 		eventbusID := vanus.NewTestID()
+		namespaceID := vanus.NewTestID()
 		ctrl.state = primitive.ServerStateRunning
 		Convey("update subscription not exist", func() {
 			subManager.EXPECT().GetSubscription(gomock.Any(), gomock.Eq(subID)).Return(nil)
@@ -181,6 +185,7 @@ func TestController_UpdateSubscription(t *testing.T) {
 			Phase:         metadata.SubscriptionPhaseStopped,
 			TriggerWorker: "test-addr",
 			EventbusID:    eventbusID,
+			NamespaceID:   namespaceID,
 			Name:          "test-name",
 			Sink:          "test-sink",
 			Protocol:      primitive.HTTPProtocol,
@@ -194,8 +199,21 @@ func TestController_UpdateSubscription(t *testing.T) {
 			request := &ctrlpb.UpdateSubscriptionRequest{
 				Id: subID.Uint64(),
 				Subscription: &ctrlpb.SubscriptionRequest{
-					EventbusId: eventbusID.Uint64(),
-					Sink:       "test-sink",
+					NamespaceId: namespaceID.Uint64(),
+					EventbusId:  eventbusID.Uint64(),
+					Sink:        "test-sink",
+				},
+			}
+			_, err := ctrl.UpdateSubscription(ctx, request)
+			So(err, ShouldNotBeNil)
+		})
+		Convey("tet update namespace fail", func() {
+			request := &ctrlpb.UpdateSubscriptionRequest{
+				Id: subID.Uint64(),
+				Subscription: &ctrlpb.SubscriptionRequest{
+					NamespaceId: vanus.NewTestID().Uint64(),
+					EventbusId:  eventbusID.Uint64(),
+					Sink:        "test-sink",
 				},
 			}
 			_, err := ctrl.UpdateSubscription(ctx, request)
@@ -205,8 +223,9 @@ func TestController_UpdateSubscription(t *testing.T) {
 			request := &ctrlpb.UpdateSubscriptionRequest{
 				Id: subID.Uint64(),
 				Subscription: &ctrlpb.SubscriptionRequest{
-					EventbusId: vanus.NewTestID().Uint64(),
-					Sink:       "test-sink",
+					NamespaceId: namespaceID.Uint64(),
+					EventbusId:  vanus.NewTestID().Uint64(),
+					Sink:        "test-sink",
 				},
 			}
 			_, err := ctrl.UpdateSubscription(ctx, request)
@@ -217,9 +236,10 @@ func TestController_UpdateSubscription(t *testing.T) {
 				request := &ctrlpb.UpdateSubscriptionRequest{
 					Id: subID.Uint64(),
 					Subscription: &ctrlpb.SubscriptionRequest{
-						EventbusId: eventbusID.Uint64(),
-						Sink:       "test-sink",
-						Protocol:   metapb.Protocol_AWS_LAMBDA,
+						NamespaceId: namespaceID.Uint64(),
+						EventbusId:  eventbusID.Uint64(),
+						Sink:        "test-sink",
+						Protocol:    metapb.Protocol_AWS_LAMBDA,
 						SinkCredential: &metapb.SinkCredential{
 							CredentialType: metapb.SinkCredential_AWS,
 						},
@@ -232,9 +252,10 @@ func TestController_UpdateSubscription(t *testing.T) {
 				request := &ctrlpb.UpdateSubscriptionRequest{
 					Id: subID.Uint64(),
 					Subscription: &ctrlpb.SubscriptionRequest{
-						EventbusId: eventbusID.Uint64(),
-						Sink:       "arn:aws:lambda:us-west-2:843378899134:function:xdltest",
-						Protocol:   metapb.Protocol_AWS_LAMBDA,
+						NamespaceId: namespaceID.Uint64(),
+						EventbusId:  eventbusID.Uint64(),
+						Sink:        "arn:aws:lambda:us-west-2:843378899134:function:xdltest",
+						Protocol:    metapb.Protocol_AWS_LAMBDA,
 						SinkCredential: &metapb.SinkCredential{
 							CredentialType: metapb.SinkCredential_PLAIN,
 						},
@@ -247,9 +268,10 @@ func TestController_UpdateSubscription(t *testing.T) {
 				request := &ctrlpb.UpdateSubscriptionRequest{
 					Id: subID.Uint64(),
 					Subscription: &ctrlpb.SubscriptionRequest{
-						EventbusId: eventbusID.Uint64(),
-						Sink:       "arn:aws:lambda:us-west-2:843378899134:function:xdltest",
-						Protocol:   metapb.Protocol_AWS_LAMBDA,
+						NamespaceId: namespaceID.Uint64(),
+						EventbusId:  eventbusID.Uint64(),
+						Sink:        "arn:aws:lambda:us-west-2:843378899134:function:xdltest",
+						Protocol:    metapb.Protocol_AWS_LAMBDA,
 						SinkCredential: &metapb.SinkCredential{
 							CredentialType: metapb.SinkCredential_AWS,
 						},
@@ -262,10 +284,11 @@ func TestController_UpdateSubscription(t *testing.T) {
 				request := &ctrlpb.UpdateSubscriptionRequest{
 					Id: subID.Uint64(),
 					Subscription: &ctrlpb.SubscriptionRequest{
-						EventbusId: eventbusID.Uint64(),
-						Name:       "test-name",
-						Sink:       "arn:aws:lambda:us-west-2:843378899134:function:xdltest",
-						Protocol:   metapb.Protocol_AWS_LAMBDA,
+						NamespaceId: namespaceID.Uint64(),
+						EventbusId:  eventbusID.Uint64(),
+						Name:        "test-name",
+						Sink:        "arn:aws:lambda:us-west-2:843378899134:function:xdltest",
+						Protocol:    metapb.Protocol_AWS_LAMBDA,
 						SinkCredential: &metapb.SinkCredential{
 							CredentialType: metapb.SinkCredential_AWS,
 							Credential: &metapb.SinkCredential_Aws{
@@ -292,9 +315,10 @@ func TestController_UpdateSubscription(t *testing.T) {
 			request := &ctrlpb.UpdateSubscriptionRequest{
 				Id: subID.Uint64(),
 				Subscription: &ctrlpb.SubscriptionRequest{
-					EventbusId: eventbusID.Uint64(),
-					Name:       "test-name",
-					Sink:       "modify-sink",
+					NamespaceId: namespaceID.Uint64(),
+					EventbusId:  eventbusID.Uint64(),
+					Name:        "test-name",
+					Sink:        "modify-sink",
 				},
 			}
 			resp, err := ctrl.UpdateSubscription(ctx, request)
@@ -308,9 +332,10 @@ func TestController_UpdateSubscription(t *testing.T) {
 			request := &ctrlpb.UpdateSubscriptionRequest{
 				Id: subID.Uint64(),
 				Subscription: &ctrlpb.SubscriptionRequest{
-					EventbusId: eventbusID.Uint64(),
-					Name:       "test-name",
-					Sink:       "test-sink",
+					NamespaceId: namespaceID.Uint64(),
+					EventbusId:  eventbusID.Uint64(),
+					Name:        "test-name",
+					Sink:        "test-sink",
 					Filters: []*metapb.Filter{
 						{
 							Exact: map[string]string{"type": "test"},
@@ -328,9 +353,10 @@ func TestController_UpdateSubscription(t *testing.T) {
 			request := &ctrlpb.UpdateSubscriptionRequest{
 				Id: subID.Uint64(),
 				Subscription: &ctrlpb.SubscriptionRequest{
-					EventbusId: eventbusID.Uint64(),
-					Name:       "test-name",
-					Sink:       "test-sink",
+					NamespaceId: namespaceID.Uint64(),
+					EventbusId:  eventbusID.Uint64(),
+					Name:        "test-name",
+					Sink:        "test-sink",
 					Transformer: &metapb.Transformer{
 						Define:   map[string]string{"k": "v"},
 						Template: "test",

internal/controller/trigger/metadata/info.go

@@ -76,6 +76,7 @@ type Subscription struct {
 	Protocol           primitive.Protocol              `json:"protocol,omitempty"`
 	ProtocolSetting    *primitive.ProtocolSetting      `json:"protocol_settings,omitempty"`
 	EventbusID         vanus.ID                        `json:"eventbus_id"`
+	NamespaceID        vanus.ID                        `json:"namespace_id"`
 	Transformer        *primitive.Transformer          `json:"transformer,omitempty"`
 	Name               string                          `json:"name"`
 	Description        string                          `json:"description"`

internal/controller/trigger/subscription/subscription.go

@@ -48,7 +48,7 @@ type Manager interface {
 	ResetOffsetByTimestamp(ctx context.Context, id vanus.ID, timestamp uint64) (info.ListOffsetInfo, error)
 	ListSubscription(ctx context.Context) []*metadata.Subscription
 	GetSubscription(ctx context.Context, id vanus.ID) *metadata.Subscription
-	GetSubscriptionByName(ctx context.Context, eventbusID vanus.ID, name string) *metadata.Subscription
+	GetSubscriptionByName(ctx context.Context, namespaceID vanus.ID, name string) *metadata.Subscription
 	AddSubscription(ctx context.Context, subscription *metadata.Subscription) error
 	UpdateSubscription(ctx context.Context, subscription *metadata.Subscription) error
 	Heartbeat(ctx context.Context, id vanus.ID, addr string, time time.Time) error
@@ -104,11 +104,11 @@ func (m *manager) ListSubscription(_ context.Context) []*metadata.Subscription {
 	return list
 }
 
-func (m *manager) GetSubscriptionByName(ctx context.Context, eventbusID vanus.ID, name string) *metadata.Subscription {
+func (m *manager) GetSubscriptionByName(ctx context.Context, namespaceID vanus.ID, name string) *metadata.Subscription {
 	m.lock.RLock()
 	defer m.lock.RUnlock()
 	for _, sub := range m.subscriptionMap {
-		if sub.EventbusID != eventbusID {
+		if sub.NamespaceID != namespaceID {
 			continue
 		}
 		if sub.Name == name {

internal/controller/trigger/validation/subscripton.go

@@ -24,6 +24,7 @@ import (
 	"google.golang.org/api/idtoken"
 	"google.golang.org/api/option"
 
+	"github.com/vanus-labs/vanus/internal/primitive/vanus"
 	"github.com/vanus-labs/vanus/pkg/errors"
 	ctrlpb "github.com/vanus-labs/vanus/proto/pkg/controller"
 	metapb "github.com/vanus-labs/vanus/proto/pkg/meta"
@@ -32,10 +33,15 @@ import (
 	"github.com/vanus-labs/vanus/internal/primitive/cel"
 	"github.com/vanus-labs/vanus/internal/primitive/transform/arg"
 	"github.com/vanus-labs/vanus/internal/primitive/transform/runtime"
-	"github.com/vanus-labs/vanus/internal/primitive/vanus"
 )
 
 func ValidateSubscriptionRequest(ctx context.Context, request *ctrlpb.SubscriptionRequest) error {
+	if request.NamespaceId == vanus.EmptyID().Uint64() {
+		return errors.ErrInvalidRequest.WithMessage("namespace is empty")
+	}
+	if request.EventbusId == vanus.EmptyID().Uint64() {
+		return errors.ErrInvalidRequest.WithMessage("eventbus is empty")
+	}
 	if err := ValidateFilterList(ctx, request.Filters); err != nil {
 		return errors.ErrInvalidRequest.WithMessage("filters is invalid").Wrap(err)
 	}
@@ -48,9 +54,6 @@ func ValidateSubscriptionRequest(ctx context.Context, request *ctrlpb.Subscripti
 	if err := validateSinkCredential(ctx, request.Sink, request.SinkCredential); err != nil {
 		return err
 	}
-	if request.EventbusId == vanus.EmptyID().Uint64() {
-		return errors.ErrInvalidRequest.WithMessage("eventbus is empty")
-	}
 	if request.Name == "" {
 		return errors.ErrInvalidRequest.WithMessage("name is empty")
 	}

internal/controller/trigger/validation/suscription_test.go

@@ -30,6 +30,8 @@ func TestSubscriptionRequestValidator(t *testing.T) {
 	ctx := context.Background()
 	Convey("multiple dialect", t, func() {
 		request := &ctrlpb.SubscriptionRequest{
+			NamespaceId: vanus.NewTestID().Uint64(),
+			EventbusId:  vanus.NewTestID().Uint64(),
 			Filters: []*metapb.Filter{{
 				Not: &metapb.Filter{
 					Exact: map[string]string{
@@ -43,8 +45,7 @@ func TestSubscriptionRequestValidator(t *testing.T) {
 	})
 	Convey("eventbus empty", t, func() {
 		request := &ctrlpb.SubscriptionRequest{
-			Sink:       "sink",
-			EventbusId: vanus.EmptyID().Uint64(),
+			Sink: "sink",
 		}
 		So(ValidateSubscriptionRequest(ctx, request), ShouldNotBeNil)
 	})

internal/convert/convert.go

@@ -40,6 +40,7 @@ func FromPbSubscriptionRequest(sub *ctrl.SubscriptionRequest) *metadata.Subscrip
 		Filters:            fromPbFilters(sub.Filters),
 		Transformer:        fromPbTransformer(sub.Transformer),
 		EventbusID:         vanus.NewIDFromUint64(sub.EventbusId),
+		NamespaceID:        vanus.NewIDFromUint64(sub.NamespaceId),
 		Name:               sub.Name,
 		Description:        sub.Description,
 	}
@@ -295,6 +296,7 @@ func ToPbSubscription(sub *metadata.Subscription, offsets info.ListOffsetInfo) *
 		Protocol:         toPbProtocol(sub.Protocol),
 		ProtocolSettings: toPbProtocolSettings(sub.ProtocolSetting),
 		EventbusId:       sub.EventbusID.Uint64(),
+		NamespaceId:      sub.NamespaceID.Uint64(),
 		Filters:          toPbFilters(sub.Filters),
 		Transformer:      ToPbTransformer(sub.Transformer),
 		Offsets:          ToPbOffsetInfos(offsets),

vsctl/command/subscription.go

@@ -96,6 +96,7 @@ func createSubscriptionCommand() *cobra.Command {
 					Sink:           sink,
 					SinkCredential: credential,
 					Protocol:       p,
+					NamespaceId:    mustGetNamespaceID(namespace).Uint64(),
 					EventbusId:     mustGetEventbusID(namespace, eventbus).Uint64(),
 					Transformer:    trans,
 					Name:           subscriptionName,
@@ -358,6 +359,7 @@ func updateSubscriptionCommand() *cobra.Command {
 					Sink:           sub.Sink,
 					SinkCredential: sub.SinkCredential,
 					Protocol:       sub.Protocol,
+					NamespaceId:    mustGetNamespaceID(namespace).Uint64(),
 					EventbusId:     mustGetEventbusID(namespace, eventbus).Uint64(),
 					Transformer:    sub.Transformer,
 					Name:           sub.Name,
@@ -577,6 +579,9 @@ func listSubscriptionCommand() *cobra.Command {
 			request := &ctrlpb.ListSubscriptionRequest{
 				Name: subscriptionName,
 			}
+			if namespace != "" {
+				request.NamespaceId = mustGetNamespaceID(namespace).Uint64()
+			}
 			if eventbus != "" {
 				request.EventbusId = mustGetEventbusID(namespace, eventbus).Uint64()
 			}