chore(service): implement private pipeline quota check

go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/instill-ai/component v0.7.1-alpha.0.20231208045032-bafec3495571
 	github.com/instill-ai/connector v0.7.0-alpha.0.20231206040111-5a57a09f2adc
 	github.com/instill-ai/operator v0.5.0-alpha.0.20231206181023-581e551939b9
-	github.com/instill-ai/protogen-go v0.3.3-alpha.0.20231207112549-0e2a5afcd9c4
+	github.com/instill-ai/protogen-go v0.3.3-alpha.0.20231210131526-67e990838339
 	github.com/instill-ai/usage-client v0.2.4-alpha.0.20231206162018-6ccbff13136b
 	github.com/instill-ai/x v0.3.0-alpha.0.20231124062833-3236165f5782
 	github.com/knadh/koanf v1.5.0

go.sum

@@ -1189,8 +1189,8 @@ github.com/instill-ai/connector v0.7.0-alpha.0.20231206040111-5a57a09f2adc h1:i3
 github.com/instill-ai/connector v0.7.0-alpha.0.20231206040111-5a57a09f2adc/go.mod h1:d4/IGbcK8YM7IqrvRRN3oMQVOk4jtFRzZGnflKY1IzI=
 github.com/instill-ai/operator v0.5.0-alpha.0.20231206181023-581e551939b9 h1:Fx6UwbFek49Kgo6cbMlnxycmrvC3TPfs8Zkd1jjF99w=
 github.com/instill-ai/operator v0.5.0-alpha.0.20231206181023-581e551939b9/go.mod h1:oUG3J+ndGK0Ebxz664FaM7Csn+qPbR0gxitlFaRXTaI=
-github.com/instill-ai/protogen-go v0.3.3-alpha.0.20231207112549-0e2a5afcd9c4 h1:SHDewY9zWZSxN2ahSMxgINnBdLIrsAqjUKplwqNm398=
-github.com/instill-ai/protogen-go v0.3.3-alpha.0.20231207112549-0e2a5afcd9c4/go.mod h1:q/YL5TZXD9nvmJ7Rih4gY3/B2HT2+GiFdxeZp9D+yE4=
+github.com/instill-ai/protogen-go v0.3.3-alpha.0.20231210131526-67e990838339 h1:Q48Mm+0i6gL4ZYMiHPddMfBQaslk83y3jmPg9T1T7IQ=
+github.com/instill-ai/protogen-go v0.3.3-alpha.0.20231210131526-67e990838339/go.mod h1:q/YL5TZXD9nvmJ7Rih4gY3/B2HT2+GiFdxeZp9D+yE4=
 github.com/instill-ai/usage-client v0.2.4-alpha.0.20231206162018-6ccbff13136b h1:YwXracXgTWxaPfIsbC3uaZFjGO0E2y1e3hK9ZUq7ipE=
 github.com/instill-ai/usage-client v0.2.4-alpha.0.20231206162018-6ccbff13136b/go.mod h1:4uTzVtcC+UVKhRaNvJc2+LFLcr/fv3vsYE5QCWu6TQ0=
 github.com/instill-ai/x v0.3.0-alpha.0.20231124062833-3236165f5782 h1:ErsJ1IkjD6Rtif0YI898fv/qllW5x9vb2fdt69EYwug=

internal/resource/resource.go

@@ -58,6 +58,7 @@ const (
 
 type Namespace struct {
 	NsType NamespaceType
+	NsID   string
 	NsUid  uuid.UUID
 }
 

pkg/middleware/interceptor.go

@@ -102,7 +102,8 @@ func InjectErrCode(err error) error {
 
 	case
 		errors.Is(err, service.ErrRateLimiting),
-		errors.Is(err, service.ErrNamespaceQuotaExceed):
+		errors.Is(err, service.ErrNamespacePrivatePipelineQuotaExceed),
+		errors.Is(err, service.ErrNamespaceTriggerQuotaExceed):
 		return status.Error(codes.ResourceExhausted, err.Error())
 
 	case

pkg/service/errors.go

@@ -6,4 +6,5 @@ var ErrNoPermission = errors.New("no permission")
 var ErrNotFound = errors.New("not found")
 var ErrUnauthenticated = errors.New("unauthenticated")
 var ErrRateLimiting = errors.New("rate limiting")
-var ErrNamespaceQuotaExceed = errors.New("namespace quota exceed")
+var ErrNamespaceTriggerQuotaExceed = errors.New("namespace trigger quota exceed")
+var ErrNamespacePrivatePipelineQuotaExceed = errors.New("namespace private pipeline quota exceed")

pkg/service/service.go

@@ -22,6 +22,8 @@
 	"go.temporal.io/api/enums/v1"
 	"go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/temporal"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/reflect/protoreflect"
@@ -317,37 +319,40 @@
 	if strings.HasPrefix(name, "users") {
 		userResp, err := s.mgmtPrivateServiceClient.GetUserAdmin(context.Background(), &mgmtPB.GetUserAdminRequest{Name: name})
 		if err != nil {
-			return "", fmt.Errorf("ConvertOwnerNameToPermalink error")
+			return "", fmt.Errorf("ConvertOwnerNameToPermalink error %w", err)
 		}
 		return fmt.Sprintf("users/%s", *userResp.User.Uid), nil
 	} else {
 		orgResp, err := s.mgmtPrivateServiceClient.GetOrganizationAdmin(context.Background(), &mgmtPB.GetOrganizationAdminRequest{Name: name})
 		if err != nil {
-			return "", fmt.Errorf("ConvertOwnerNameToPermalink error")
+			return "", fmt.Errorf("ConvertOwnerNameToPermalink error %w", err)
 		}
 		return fmt.Sprintf("organizations/%s", orgResp.Organization.Uid), nil
 	}
 }
 
 func (s *service) GetRscNamespaceAndNameID(path string) (resource.Namespace, string, error) {
 
+	fmt.Println(path)
 	splits := strings.Split(path, "/")
 	if len(splits) < 2 {
 		return resource.Namespace{}, "", fmt.Errorf("namespace error")
 	}
 	uidStr, err := s.ConvertOwnerNameToPermalink(fmt.Sprintf("%s/%s", splits[0], splits[1]))
 
 	if err != nil {
-		return resource.Namespace{}, "", fmt.Errorf("namespace error")
+		return resource.Namespace{}, "", fmt.Errorf("namespace error %w", err)
 	}
 	if len(splits) < 4 {
 		return resource.Namespace{
 			NsType: resource.NamespaceType(splits[0]),
+			NsID:   splits[1],
 			NsUid:  uuid.FromStringOrNil(strings.Split(uidStr, "/")[1]),
 		}, "", nil
 	}
 	return resource.Namespace{
 		NsType: resource.NamespaceType(splits[0]),
+		NsID:   splits[1],
 		NsUid:  uuid.FromStringOrNil(strings.Split(uidStr, "/")[1]),
 	}, splits[3], nil
 }
@@ -364,11 +369,13 @@
 	if len(splits) < 4 {
 		return resource.Namespace{
 			NsType: resource.NamespaceType(splits[0]),
+			NsID:   splits[1],
 			NsUid:  uuid.FromStringOrNil(strings.Split(uidStr, "/")[1]),
 		}, uuid.Nil, nil
 	}
 	return resource.Namespace{
 		NsType: resource.NamespaceType(splits[0]),
+		NsID:   splits[1],
 		NsUid:  uuid.FromStringOrNil(strings.Split(uidStr, "/")[1]),
 	}, uuid.FromStringOrNil(splits[3]), nil
 }
@@ -457,6 +464,29 @@
 	return s.DBToPBPipeline(ctx, dbPipeline, view)
 }
 
+func (s *service) checkPrivatePipelineQuota(ctx context.Context, ns resource.Namespace, dbPipeline *datamodel.Pipeline, quota int) error {
+
+	if dbPipeline.Permission.Users["*/*"].Enabled {
+		return nil
+	}
+	privateCount := 0
+	// TODO: optimize this
+	pipelines, _, _, err := s.repository.ListPipelinesAdmin(ctx, 100, "", true, filtering.Filter{}, false)
+	if err != nil {
+		return err
+	}
+	for _, pipeline := range pipelines {
+		if !pipeline.Permission.Users["*/*"].Enabled {
+			privateCount += 1
+		}
+	}
+	if privateCount >= quota {
+		return ErrNamespacePrivatePipelineQuotaExceed
+	}
+
+	return nil
+}
+
 func (s *service) CreateNamespacePipeline(ctx context.Context, ns resource.Namespace, authUser *AuthUser, pbPipeline *pipelinePB.Pipeline) (*pipelinePB.Pipeline, error) {
 
 	ownerPermalink := ns.String()
@@ -477,11 +507,33 @@
 	}
 
 	dbPipeline, err := s.PBToDBPipeline(ctx, pbPipeline)
-
 	if err != nil {
 		return nil, err
 	}
 
+	quota := -1
+	resp, err := s.mgmtPrivateServiceClient.GetOrganizationSubscriptionAdmin(ctx,
+		&mgmtPB.GetOrganizationSubscriptionAdminRequest{Parent: fmt.Sprintf("%s/%s", ns.NsType, ns.NsID)},
+	)
+	if err != nil {
+		s, ok := status.FromError(err)
+		if !ok {
+			return nil, err
+		}
+		if s.Code() != codes.Unimplemented {
+			return nil, err
+		}
+	} else {
+		quota = int(resp.Subscription.Quota.PrivatePipeline.Quota)
+	}
+
+	if quota > -1 {
+		err = s.checkPrivatePipelineQuota(ctx, ns, dbPipeline, quota)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	if dbPipeline.ShareCode == "" {
 		dbPipeline.ShareCode = GenerateShareCode()
 	}
@@ -601,18 +653,18 @@
 
 	ownerPermalink := ns.String()
 
-	dbPipelineToCreate, err := s.PBToDBPipeline(ctx, toUpdPipeline)
+	dbPipelineToUpdate, err := s.PBToDBPipeline(ctx, toUpdPipeline)
 	if err != nil {
 		return nil, ErrNotFound
 	}
 
-	if granted, err := s.aclClient.CheckPermission("pipeline", dbPipelineToCreate.UID, authUser.GetACLType(), authUser.UID, s.getCode(ctx), "reader"); err != nil {
+	if granted, err := s.aclClient.CheckPermission("pipeline", dbPipelineToUpdate.UID, authUser.GetACLType(), authUser.UID, s.getCode(ctx), "reader"); err != nil {
 		return nil, err
 	} else if !granted {
 		return nil, ErrNotFound
 	}
 
-	if granted, err := s.aclClient.CheckPermission("pipeline", dbPipelineToCreate.UID, authUser.GetACLType(), authUser.UID, s.getCode(ctx), "admin"); err != nil {
+	if granted, err := s.aclClient.CheckPermission("pipeline", dbPipelineToUpdate.UID, authUser.GetACLType(), authUser.UID, s.getCode(ctx), "admin"); err != nil {
 		return nil, err
 	} else if !granted {
 		return nil, ErrNoPermission
@@ -625,10 +677,33 @@
 	}
 
 	if existingPipeline.ShareCode == "" {
-		dbPipelineToCreate.ShareCode = GenerateShareCode()
+		dbPipelineToUpdate.ShareCode = GenerateShareCode()
+	}
+
+	quota := -1
+	resp, err := s.mgmtPrivateServiceClient.GetOrganizationSubscriptionAdmin(ctx,
+		&mgmtPB.GetOrganizationSubscriptionAdminRequest{Parent: fmt.Sprintf("%s/%s", ns.NsType, ns.NsID)},
+	)
+	if err != nil {
+		s, ok := status.FromError(err)
+		if !ok {
+			return nil, err
+		}
+		if s.Code() != codes.Unimplemented {
+			return nil, err
+		}
+	} else {
+		quota = int(resp.Subscription.Quota.PrivatePipeline.Quota)
+	}
+
+	if quota > -1 {
+		err = s.checkPrivatePipelineQuota(ctx, ns, dbPipelineToUpdate, quota)
+		if err != nil {
+			return nil, err
+		}
 	}
 
-	if err := s.repository.UpdateNamespacePipelineByID(ctx, ownerPermalink, id, dbPipelineToCreate); err != nil {
+	if err := s.repository.UpdateNamespacePipelineByID(ctx, ownerPermalink, id, dbPipelineToUpdate); err != nil {
 		return nil, err
 	}
 
@@ -786,7 +861,7 @@
 		if !errors.Is(err, redis.Nil) {
 			requestLeft, _ := strconv.ParseInt(value, 10, 64)
 			if requestLeft <= 0 {
-				return ErrNamespaceQuotaExceed
+				return ErrNamespaceTriggerQuotaExceed
 			} else {
 				_ = s.redisClient.Decr(context.Background(), fmt.Sprintf("namespace_quota_limit:%s:%s", n, ns.NsUid))
 			}