fix: aggregate collection-mode widget pipes correctly (IN-1195)

Addresses Copilot/Cursor Bugbot findings on PR #4348:

- activities_filtered / activities_filtered_retention: collection requests
  now read from activityRelations_deduplicated_cleaned_bucket_union instead
  of the single-bucket activityRelations_bucket_routing pipe. The routing
  pipe defaults to bucket 0 when bucketId isn't supplied (only injected for
  single-project requests), so collection-scoped segments in other hash
  buckets were silently dropped. Project-scope path is unchanged.

- package_metrics: added a per-project aggregation step before the
  existing collapse, so collection mode sums each project's own
  max/latest metrics instead of taking a single max() across all
  projects' raw rows (which returned only the largest project's values).

- search_volume: collection mode now groups by dataTimestamp and sums
  volume into one series; project mode is unchanged. Previously collection
  requests returned duplicate dates with per-project volumes.

- member_roles: added a collectionSlug branch scoped via
  segments_filtered_by_collection. Previously only `project` was checked,
  so collection-scoped contributors_leaderboard requests joined roles from
  every project in the system, not just the collection.

Signed-off-by: Gašper Grom <gasper.grom@gmail.com>

Author: gaspergrom

services/libs/tinybird/pipes/activities_filtered.pipe

@@ -24,7 +24,10 @@ NODE activities_filtered_bucket_routing
 SQL >
     %
     SELECT activityId as id, timestamp, type, platform, memberId, organizationId, segmentId
-    FROM activityRelations_bucket_routing a
+    FROM
+        {% if defined(collectionSlug) %} activityRelations_deduplicated_cleaned_bucket_union
+        {% else %} activityRelations_bucket_routing
+        {% end %} as a
     where
         {% if defined(collectionSlug) %}
             segmentId IN (SELECT segmentId FROM segments_filtered_by_collection)

services/libs/tinybird/pipes/activities_filtered_retention.pipe

@@ -23,7 +23,10 @@ NODE activities_filtered_retention_bucket_routing
 SQL >
     %
     SELECT activityId as id, timestamp, type, platform, memberId, organizationId, segmentId
-    FROM activityRelations_bucket_routing a
+    FROM
+        {% if defined(collectionSlug) %} activityRelations_deduplicated_cleaned_bucket_union
+        {% else %} activityRelations_bucket_routing
+        {% end %} as a
     where
         {% if defined(collectionSlug) %}
             segmentId IN (SELECT segmentId FROM segments_filtered_by_collection)

services/libs/tinybird/pipes/member_roles.pipe

@@ -5,7 +5,9 @@ SQL >
     FROM maintainers_roles_copy_ds
     WHERE
         1 = 1
-        {% if defined(project) %}
+        {% if defined(collectionSlug) %}
+            AND insightsProjectId IN (SELECT insightsProjectId FROM segments_filtered_by_collection)
+        {% elif defined(project) %}
             AND insightsProjectId = (SELECT insightsProjectId FROM segments_filtered)
         {% end %}
         {% if defined(repos) %}

services/libs/tinybird/pipes/package_metrics.pipe

@@ -144,11 +144,12 @@ SQL >
         ) package_downloads_timeseries_bounds
     WHERE "startDate" >= actual_start_date AND "startDate" < actual_end_date
 
-NODE package_downloads_initial_aggregation
+NODE package_downloads_per_project_aggregation
 SQL >
     %
     {% if defined(granularity) %}
         SELECT
+            p.insightsProjectId,
             p.date,
             sum(p.downloadsCount) as downloadsCount,
             sum(p.dockerDownloadsCount) as "dockerDownloadsCount",
@@ -159,6 +160,7 @@ SQL >
         group by p.insightsProjectId, p.date
     {% else %}
         SELECT
+            p.insightsProjectId,
             max(p.downloadsCount) as downloadsCount,
             max(p.dockerDownloadsCount) as "dockerDownloadsCount",
             max(p.dockerDependentsCount) as "dockerDependentsCount",
@@ -168,6 +170,29 @@ SQL >
         group by p.insightsProjectId
     {% end %}
 
+NODE package_downloads_initial_aggregation
+SQL >
+    %
+    {% if defined(granularity) %}
+        SELECT
+            date,
+            sum(downloadsCount) as downloadsCount,
+            sum(dockerDownloadsCount) as "dockerDownloadsCount",
+            sum(dockerDependentsCount) as "dockerDependentsCount",
+            sum(dependentPackagesCount) as "dependentPackagesCount",
+            sum(dependentReposCount) as "dependentReposCount"
+        FROM package_downloads_per_project_aggregation
+        group by date
+    {% else %}
+        SELECT
+            sum(downloadsCount) as downloadsCount,
+            sum(dockerDownloadsCount) as "dockerDownloadsCount",
+            sum(dockerDependentsCount) as "dockerDependentsCount",
+            sum(dependentPackagesCount) as "dependentPackagesCount",
+            sum(dependentReposCount) as "dependentReposCount"
+        FROM package_downloads_per_project_aggregation
+    {% end %}
+
 NODE package_downloads_timeseries_merge
 SQL >
     %

services/libs/tinybird/pipes/search_volume.pipe

@@ -9,17 +9,14 @@ DESCRIPTION >
     - `collectionSlug`: Collection slug (e.g., 'cncf') - inherited from `segments_filtered_by_collection`. Required unless `project` is given.
     - `startDate`: Optional DateTime filter for search volume data after timestamp (e.g., '2024-01-01 00:00:00')
     - `endDate`: Optional DateTime filter for search volume data before timestamp (e.g., '2024-12-31 23:59:59')
-    - Response: `insightsProjectId`, `project`, `dataTimestamp` (formatted as YYYY-MM-DD), `volume`, `updatedAt`
+    - Response: `insightsProjectId`, `project`, `dataTimestamp` (formatted as YYYY-MM-DD), `volume`, `updatedAt`.
+    When `collectionSlug` is given, rows are aggregated to one row per `dataTimestamp` with `volume` summed
+    across every project in the collection; `insightsProjectId`/`project`/`updatedAt` are empty in that mode.
 
-NODE searchVolume_pipe
+NODE searchVolume_filtered
 SQL >
     %
-    SELECT
-        insightsProjectId,
-        project,
-        formatDateTime(dataTimestamp, '%Y-%m-%d') AS dataTimestamp,
-        volume,
-        updatedAt
+    SELECT insightsProjectId, project, dataTimestamp, volume, updatedAt
     FROM searchVolume FINAL
     WHERE
         {% if defined(collectionSlug) %}
@@ -48,3 +45,25 @@ SQL >
                 }}
             )
         {% end %}
+
+NODE searchVolume_pipe
+SQL >
+    %
+    {% if defined(collectionSlug) %}
+        SELECT
+            toUUID('00000000-0000-0000-0000-000000000000') AS insightsProjectId,
+            '' AS project,
+            formatDateTime(dataTimestamp, '%Y-%m-%d') AS dataTimestamp,
+            sum(volume) AS volume,
+            max(updatedAt) AS updatedAt
+        FROM searchVolume_filtered
+        GROUP BY formatDateTime(dataTimestamp, '%Y-%m-%d')
+    {% else %}
+        SELECT
+            insightsProjectId,
+            project,
+            formatDateTime(dataTimestamp, '%Y-%m-%d') AS dataTimestamp,
+            volume,
+            updatedAt
+        FROM searchVolume_filtered
+    {% end %}