Fixes 4932: Add list module streams for a snapshots

compose_files/pulp/docker-compose.yml

@@ -1,15 +1,16 @@
-version: '3'
+version: "3"
 services:
   postgres:
     image: "docker.io/library/postgres:13"
+    platform: linux/amd64
     ports:
       - "5434:5432"
     environment:
       POSTGRES_USER: pulp
       POSTGRES_PASSWORD: password
       POSTGRES_DB: pulp
-      POSTGRES_INITDB_ARGS: '--auth-host=scram-sha-256'
-      POSTGRES_HOST_AUTH_METHOD: 'scram-sha-256'
+      POSTGRES_INITDB_ARGS: "--auth-host=scram-sha-256"
+      POSTGRES_HOST_AUTH_METHOD: "scram-sha-256"
     volumes:
       - "pg_data:/var/lib/postgresql"
       - "./assets/postgres/passwd:/etc/passwd:Z"
@@ -22,6 +23,7 @@ services:
 
   migration_service:
     image: "quay.io/redhat-services-prod/pulp-services-tenant/pulp:latest"
+    platform: linux/amd64
     depends_on:
       postgres:
         condition: service_healthy
@@ -30,10 +32,11 @@ services:
     volumes:
       - "./assets/settings.py:/etc/pulp/settings.py:z"
       - "./assets/certs:/etc/pulp/certs:z"
-      - "pulp:/var/lib/pulp"      
+      - "pulp:/var/lib/pulp"
 
   set_init_password_service:
-    image:  "quay.io/redhat-services-prod/pulp-services-tenant/pulp:latest"
+    image: "quay.io/redhat-services-prod/pulp-services-tenant/pulp:latest"
+    platform: linux/amd64
     command: set_init_password.sh
     depends_on:
       migration_service:
@@ -51,13 +54,14 @@ services:
       - "redis_data:/data"
     restart: always
     healthcheck:
-      test: [ "CMD", "redis-cli", "--raw", "incr", "ping" ]
+      test: ["CMD", "redis-cli", "--raw", "incr", "ping"]
 
   pulp_api:
     image: "quay.io/redhat-services-prod/pulp-services-tenant/pulp:latest"
+    platform: linux/amd64
     deploy:
       replicas: 1
-    command: [ 'pulp-api' ]
+    command: ["pulp-api"]
     depends_on:
       migration_service:
         condition: service_completed_successfully
@@ -77,9 +81,10 @@ services:
 
   pulp_content:
     image: "quay.io/redhat-services-prod/pulp-services-tenant/pulp:latest"
+    platform: linux/amd64
     deploy:
       replicas: 1
-    command: [ 'pulp-content' ]
+    command: ["pulp-content"]
     depends_on:
       migration_service:
         condition: service_completed_successfully
@@ -98,7 +103,8 @@ services:
 
   pulp_web:
     image: "pulp/pulp-web:latest"
-    command: [ '/usr/bin/nginx.sh' ]
+    platform: linux/amd64
+    command: ["/usr/bin/nginx.sh"]
     depends_on:
       migration_service:
         condition: service_completed_successfully
@@ -113,9 +119,10 @@ services:
 
   pulp_worker:
     image: "quay.io/redhat-services-prod/pulp-services-tenant/pulp:latest"
+    platform: linux/amd64
     deploy:
       replicas: 1
-    command: [ 'pulp-worker' ]
+    command: ["pulp-worker"]
     depends_on:
       migration_service:
         condition: service_completed_successfully

pkg/tangy/interface.go

@@ -56,6 +56,7 @@ type Tangy interface {
 	RpmRepositoryVersionPackageGroupSearch(ctx context.Context, hrefs []string, search string, limit int) ([]RpmPackageGroupSearch, error)
 	RpmRepositoryVersionEnvironmentSearch(ctx context.Context, hrefs []string, search string, limit int) ([]RpmEnvironmentSearch, error)
 	RpmRepositoryVersionPackageList(ctx context.Context, hrefs []string, filterOpts RpmListFilters, pageOpts PageOptions) ([]RpmListItem, int, error)
+	RpmRepositoryVersionModuleStreams(ctx context.Context, hrefs []string, rpmNames []string, search string, pageOpts PageOptions) ([]ModuleStreams, int, error)
 	RpmRepositoryVersionErrataList(ctx context.Context, hrefs []string, filterOpts ErrataListFilters, pageOpts PageOptions) ([]ErrataListItem, int, error)
 	Close()
 }

pkg/tangy/rpm.go

@@ -47,6 +47,11 @@ type RpmListItem struct {
 	Summary string // The summary of the rpm
 }
 
+type ModuleStreams struct {
+	ModuleName string   `json:"module_name"` // Snapshot UUIDs to find modules and streams for
+	Streams    []string `json:"streams"`     // Package names to filter the above list by
+}
+
 type ErrataListItem struct {
 	Id              string
 	ErrataId        string
@@ -353,6 +358,92 @@ func (t *tangyImpl) RpmRepositoryVersionErrataList(ctx context.Context, hrefs []
 	return errata, countTotal, nil
 }
 
+// RpmRepositoryVersionModuleStreams List Modules streams within a repository version, with pagination, search and an optional name filter
+func (t *tangyImpl) RpmRepositoryVersionModuleStreams(ctx context.Context, hrefs []string, rpmNames []string, search string, pageOpts PageOptions) ([]ModuleStreams, int, error) {
+	if len(hrefs) == 0 {
+		return []ModuleStreams{}, 0, nil
+	}
+
+	conn, err := t.pool.Acquire(ctx)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer conn.Release()
+
+	if pageOpts.Limit == 0 {
+		pageOpts.Limit = DefaultLimit
+	}
+
+	repoVerMap, err := parseRepositoryVersionHrefsMap(hrefs)
+	if err != nil {
+		return nil, 0, fmt.Errorf("error parsing repository version hrefs: %w", err)
+	}
+
+	orderBy := ""
+
+	switch true {
+	case strings.Contains(pageOpts.SortBy, "streams"):
+		orderBy = "streams"
+	default:
+		orderBy = "module_name"
+	}
+
+	if strings.Contains(pageOpts.SortBy, "desc") {
+		orderBy += " DESC"
+	} else {
+		orderBy += " ASC"
+	}
+
+	args := pgx.NamedArgs{
+		"nameFilter": "%" + search + "%",
+		"limit":      pageOpts.Limit,
+		"offset":     pageOpts.Offset,
+		"rpm_names":  rpmNames,
+	}
+
+	countQueryOpen := `SELECT COUNT(*) FROM (Select distinct(rp.name) FROM rpm_modulemd rp 
+	INNER JOIN rpm_modulemd_packages rmp on rmp.modulemd_id = rp.content_ptr_id
+	INNER JOIN rpm_package pack on pack.content_ptr_id = rmp.package_id `
+	innerUnion := contentIdsInVersions(repoVerMap, &args)
+
+	rpmNameFilter := ""
+
+	if len(rpmNames) > 0 {
+		rpmNameFilter = " AND pack.name = ANY(@rpm_names)"
+	}
+
+	filter := rpmNameFilter + " AND rp.name ILIKE CONCAT( '%', @nameFilter::text, '%') GROUP BY rp.name"
+
+	var countTotal int
+
+	err = conn.QueryRow(ctx, countQueryOpen+innerUnion+filter+")", args).Scan(&countTotal)
+
+	if err != nil {
+		if err == pgx.ErrNoRows {
+			return []ModuleStreams{}, 0, nil
+		}
+		return nil, 0, err
+	}
+
+	query := `SELECT rp.name as module_name, ARRAY_AGG(distinct rp.stream) as streams FROM rpm_modulemd rp 
+	INNER JOIN rpm_modulemd_packages rmp on rmp.modulemd_id = rp.content_ptr_id
+	INNER JOIN rpm_package pack on pack.content_ptr_id = rmp.package_id `
+
+	rows, err := conn.Query(ctx, query+innerUnion+filter+" ORDER BY "+orderBy+" LIMIT @limit OFFSET @offset", args)
+
+	if err != nil {
+		return nil, 0, err
+	}
+
+	moduleStreams, err := pgx.CollectRows(rows, pgx.RowToStructByName[ModuleStreams])
+
+	if err != nil {
+		return nil, 0, err
+	}
+
+	return moduleStreams, countTotal, nil
+}
+
 // RpmRepositoryVersionPackageList List RPMs within a repository version, with pagination, and an optional name filter
 func (t *tangyImpl) RpmRepositoryVersionPackageList(ctx context.Context, hrefs []string, filterOpts RpmListFilters, pageOpts PageOptions) ([]RpmListItem, int, error) {
 	if len(hrefs) == 0 {