chore(service): update subscription endpoints

Author: donch1989

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0
 	github.com/iancoleman/strcase v0.2.0
 	github.com/influxdata/influxdb-client-go/v2 v2.12.3
-	github.com/instill-ai/protogen-go v0.3.3-alpha.0.20231208161503-706306fd242e
+	github.com/instill-ai/protogen-go v0.3.3-alpha.0.20231210131526-67e990838339
 	github.com/instill-ai/usage-client v0.2.4-alpha.0.20231206035716-4c05f872df97
 	github.com/instill-ai/x v0.3.0-alpha
 	github.com/knadh/koanf v1.4.4

go.sum

@@ -1105,8 +1105,8 @@ github.com/influxdata/line-protocol/v2 v2.0.0-20210312151457-c52fdecb625a/go.mod
 github.com/influxdata/line-protocol/v2 v2.1.0/go.mod h1:QKw43hdUBg3GTk2iC3iyCxksNj7PX9aUSeYOYE/ceHY=
 github.com/influxdata/line-protocol/v2 v2.2.1 h1:EAPkqJ9Km4uAxtMRgUubJyqAr6zgWM0dznKMLRauQRE=
 github.com/influxdata/line-protocol/v2 v2.2.1/go.mod h1:DmB3Cnh+3oxmG6LOBIxce4oaL4CPj3OmMPgvauXh+tM=
-github.com/instill-ai/protogen-go v0.3.3-alpha.0.20231208161503-706306fd242e h1:dnZvLQnuUqpAozwj59sQRpeTF4tg+qeIkLvHKuMc788=
-github.com/instill-ai/protogen-go v0.3.3-alpha.0.20231208161503-706306fd242e/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.20231206035716-4c05f872df97 h1:WycXqzJP1ihjJwrkxlNP2TQc1DSUxUtfl/PtCpLBa3Y=
 github.com/instill-ai/usage-client v0.2.4-alpha.0.20231206035716-4c05f872df97/go.mod h1:Da8RdKakfxy1iNdvI5FSTcL1lSDtda+b9jOgOEEO68E=
 github.com/instill-ai/x v0.3.0-alpha h1:z9fedROOG2dVHhswBfVwU/hzHuq8/JKSUON7inF+FH8=

pkg/service/service.go

@@ -10,6 +10,7 @@ import (
 	"github.com/redis/go-redis/v9"
 	"go.einride.tech/aip/filtering"
 	"golang.org/x/crypto/bcrypt"
+	"google.golang.org/grpc/metadata"
 
 	"github.com/instill-ai/mgmt-backend/internal/resource"
 	"github.com/instill-ai/mgmt-backend/pkg/acl"
@@ -65,6 +66,8 @@ type Service interface {
 	CheckUserPassword(ctx context.Context, uid uuid.UUID, password string) error
 	UpdateUserPassword(ctx context.Context, uid uuid.UUID, newPassword string) error
 
+	ListUserPipelines(ctx context.Context, id string) ([]*pipelinePB.Pipeline, error)
+	ListOrganizationPipelines(ctx context.Context, id string) ([]*pipelinePB.Pipeline, error)
 	ListPipelineTriggerRecords(ctx context.Context, owner *mgmtPB.User, pageSize int64, pageToken string, filter filtering.Filter) ([]*mgmtPB.PipelineTriggerRecord, int64, string, error)
 	ListPipelineTriggerTableRecords(ctx context.Context, owner *mgmtPB.User, pageSize int64, pageToken string, filter filtering.Filter) ([]*mgmtPB.PipelineTriggerTableRecord, int64, string, error)
 	ListPipelineTriggerChartRecords(ctx context.Context, owner *mgmtPB.User, aggregationWindow int64, filter filtering.Filter) ([]*mgmtPB.PipelineTriggerChartRecord, error)
@@ -802,3 +805,58 @@ func (s *service) DeleteOrganizationMembership(ctx context.Context, ctxUserUID u
 	}
 	return nil
 }
+func (s *service) ListUserPipelines(ctx context.Context, id string) ([]*pipelinePB.Pipeline, error) {
+
+	pageToken := ""
+	pageSize := int32(100)
+
+	pipelines := []*pipelinePB.Pipeline{}
+	for {
+		resp, err := s.pipelinePublicServiceClient.ListUserPipelines(
+			metadata.AppendToOutgoingContext(ctx, "Jwt-Sub", resource.GetRequestSingleHeader(ctx, constant.HeaderUserUIDKey)),
+			&pipelinePB.ListUserPipelinesRequest{
+				PageSize:  &pageSize,
+				PageToken: &pageToken,
+				Parent:    fmt.Sprintf("users/%s", id),
+			},
+		)
+		if err != nil {
+			return nil, err
+		}
+		pipelines = append(pipelines, resp.Pipelines...)
+		pageToken = resp.NextPageToken
+		if pageToken == "" {
+			break
+		}
+	}
+
+	return pipelines, nil
+}
+
+func (s *service) ListOrganizationPipelines(ctx context.Context, id string) ([]*pipelinePB.Pipeline, error) {
+
+	pageToken := ""
+	pageSize := int32(100)
+
+	pipelines := []*pipelinePB.Pipeline{}
+	for {
+		resp, err := s.pipelinePublicServiceClient.ListOrganizationPipelines(
+			metadata.AppendToOutgoingContext(ctx, "Jwt-Sub", resource.GetRequestSingleHeader(ctx, constant.HeaderUserUIDKey)),
+			&pipelinePB.ListOrganizationPipelinesRequest{
+				PageSize:  &pageSize,
+				PageToken: &pageToken,
+				Parent:    fmt.Sprintf("organizations/%s", id),
+			},
+		)
+		if err != nil {
+			return nil, err
+		}
+		pipelines = append(pipelines, resp.Pipelines...)
+		pageToken = resp.NextPageToken
+		if pageToken == "" {
+			break
+		}
+	}
+
+	return pipelines, nil
+}