feat: add sink dashboard metrics (CM-811)

services/libs/tinybird/pipes/cdp_dashboard_metrics_total_sink.pipe

@@ -0,0 +1,59 @@
+DESCRIPTION >
+    Global metrics for activities, organizations, and members used for CDP dashboard.
+    Uses cdp_member_segment_aggregates_MV and cdp_organization_segment_aggregates_MV.
+    "Last30Days" is based on lastActive >= now() - 30 days.
+
+NODE cdpDashboardMetricsTotal
+SQL >
+    SELECT
+        ms.activitiesTotal,
+        ms.activitiesLast30Days,
+        ms.membersTotal,
+        ms.membersLast30Days,
+        os.organizationsTotal,
+        os.organizationsLast30Days
+    FROM
+    (
+        -- member-based global metrics (single scan over cdp_member_segment_aggregates_MV)
+        SELECT
+            sum(activityCount)                                                      AS activitiesTotal,
+            sumIf(activityCount, lastActive >= now() - INTERVAL 30 DAY)            AS activitiesLast30Days,
+            uniqCombined(memberId)                                                 AS membersTotal,
+            uniqCombinedIf(memberId, lastActive >= now() - INTERVAL 30 DAY)       AS membersLast30Days
+        FROM
+        (
+            -- finalize AggregateFunction states per member
+            SELECT
+                memberId,
+                countMerge(activityCountState) AS activityCount,
+                maxMerge(lastActiveState)     AS lastActive
+            FROM cdp_member_segment_aggregates_MV
+            GROUP BY
+                memberId
+        ) AS m
+    ) AS ms
+    CROSS JOIN
+    (
+        -- organization-based global metrics (single scan over cdp_organization_segment_aggregates_MV)
+        SELECT
+            uniqCombined(organizationId)                                           AS organizationsTotal,
+            uniqCombinedIf(organizationId, lastActive >= now() - INTERVAL 30 DAY) AS organizationsLast30Days
+        FROM
+        (
+            -- finalize AggregateFunction states per organization
+            SELECT
+                organizationId,
+                maxMerge(lastActiveState) AS lastActive
+            FROM cdp_organization_segment_aggregates_MV
+            GROUP BY
+                organizationId
+        ) AS o
+    ) AS os
+
+TYPE SINK
+EXPORT_SERVICE kafka
+EXPORT_CONNECTION_NAME lfx-oracle-kafka-streaming
+EXPORT_SCHEDULE 0 9 * * *
+EXPORT_FORMAT json
+EXPORT_STRATEGY @new
+EXPORT_KAFKA_TOPIC cdp_dashboard_metrics_total_sink