Filter events & analytics by metadata

apps/web/app/(ee)/api/events/route.ts

@@ -10,6 +10,7 @@ import { eventsQuerySchema } from "@/lib/zod/schemas/analytics";
 import { Folder, Link } from "@dub/prisma/client";
 import { NextResponse } from "next/server";
 
+// GET /api/events
 export const GET = withWorkspace(
   async ({ searchParams, workspace, session }) => {
     throwIfClicksUsageExceeded(workspace);

apps/web/lib/analytics/get-analytics.ts

@@ -13,6 +13,7 @@ import {
   SINGULAR_ANALYTICS_ENDPOINTS,
 } from "./constants";
 import { AnalyticsFilters } from "./types";
+import { parseFiltersFromQuery } from "./utils/analytics-query-parser";
 import { getStartEndDates } from "./utils/get-start-end-dates";
 
 // Fetch data for /api/analytics
@@ -32,6 +33,7 @@ export const getAnalytics = async (params: AnalyticsFilters) => {
     timezone = "UTC",
     isDeprecatedClicksEndpoint = false,
     dataAvailableFrom,
+    query,
   } = params;
 
   const tagIds = combineTagIds(params);
@@ -104,6 +106,8 @@ export const getAnalytics = async (params: AnalyticsFilters) => {
         : analyticsResponse[groupBy],
   });
 
+  const filters = parseFiltersFromQuery(query);
+
   const response = await pipe({
     ...params,
     ...(UTM_TAGS_PLURAL_LIST.includes(groupBy)
@@ -119,6 +123,7 @@ export const getAnalytics = async (params: AnalyticsFilters) => {
     timezone,
     country,
     region,
+    filters: filters ? JSON.stringify(filters) : undefined,
   });
 
   if (groupBy === "count") {

apps/web/lib/analytics/get-events.ts

@@ -22,6 +22,7 @@ import {
   saleEventSchemaTBEndpoint,
 } from "../zod/schemas/sales";
 import { EventsFilters } from "./types";
+import { parseFiltersFromQuery } from "./utils/analytics-query-parser";
 import { getStartEndDates } from "./utils/get-start-end-dates";
 
 // Fetch data for /api/events
@@ -39,6 +40,7 @@ export const getEvents = async (params: EventsFilters) => {
     order,
     sortOrder,
     dataAvailableFrom,
+    query,
   } = params;
 
   const { startDate, endDate } = getStartEndDates({
@@ -78,6 +80,8 @@ export const getEvents = async (params: EventsFilters) => {
       }[eventType] ?? clickEventSchemaTBEndpoint,
   });
 
+  const filters = parseFiltersFromQuery(query);
+
   const response = await pipe({
     ...params,
     eventType,
@@ -89,6 +93,7 @@ export const getEvents = async (params: EventsFilters) => {
     offset: (params.page - 1) * params.limit,
     start: startDate.toISOString().replace("T", " ").replace("Z", ""),
     end: endDate.toISOString().replace("T", " ").replace("Z", ""),
+    filters: filters ? JSON.stringify(filters) : undefined,
   });
 
   const [linksMap, customersMap] = await Promise.all([

apps/web/lib/analytics/utils/analytics-query-parser.ts

@@ -0,0 +1,39 @@
+import { z } from "zod";
+import { EventsFilters } from "../types";
+
+interface Filter {
+  operand: string;
+  operator: "equals"; // can be expanded later
+  value: string;
+}
+
+export const parseFiltersFromQuery = (query: EventsFilters["query"]) => {
+  if (!query) {
+    return undefined;
+  }
+
+  const result = z
+    .object({
+      metadata: z.record(z.string(), z.string()),
+    })
+    .safeParse(query);
+
+  if (!result.success) {
+    console.log(`Ignoring the invalid query ${JSON.stringify(query)}`);
+    return undefined;
+  }
+
+  const filters: Filter[] = Object.entries(result.data.metadata)
+    .slice(0, 1)
+    .map(([key, value]) => {
+      return {
+        operand: key,
+        operator: "equals",
+        value,
+      };
+    });
+
+  console.log("filters", filters);
+
+  return filters;
+};

apps/web/lib/zod/schemas/analytics.ts

@@ -226,6 +226,26 @@ export const analyticsQuerySchema = z
       .describe(
         "Filter for root domains. If true, filter for domains only. If false, filter for links only. If undefined, return both.",
       ),
+    query: z
+      .string()
+      .transform((v) => {
+        try {
+          return JSON.parse(v);
+        } catch (e) {
+          return undefined;
+        }
+      })
+      .optional()
+      .describe(
+        "Filter by event metadata key-value pairs. Format: key=value. Only available for lead and sale events.",
+      )
+      .openapi({
+        example: {
+          metadata: {
+            eventType: "project_transfer",
+          },
+        },
+      }),
   })
   .merge(utmTagsSchema);
 
@@ -254,6 +274,7 @@ export const analyticsFilterTB = z
       .optional()
       .describe("The folder IDs to retrieve analytics for."),
     isMegaFolder: z.boolean().optional(),
+    filters: z.any(),
   })
   .merge(
     analyticsQuerySchema.pick({

packages/tinybird/pipes/v2_events.pipe

@@ -133,6 +133,15 @@ SQL >
             AND url LIKE concat('%utm_content=', encodeURLFormComponent({{ String(utm_content) }}), '%')
         {% end %}
         {% if defined(url) %} AND url = {{ url }} {% end %}
+
+        {% if defined(filters) %}
+            {% for item in JSON(filters, '[]') %}
+                {% if item.get('operator', '') == 'equals' %}
+                    AND JSONExtractString(metadata, {{ item.get('operand', '') }}) = {{ item.get('value', '') }}
+                {% end %}
+            {% end %}
+        {% end %}
+
     ORDER BY timestamp {% if order == 'asc' %} ASC {% else %} DESC {% end %}
     LIMIT {{ Int32(limit, 100) }}
     {% if defined(offset) %} OFFSET {{ Int32(offset, 0) }} {% end %}
@@ -174,6 +183,15 @@ SQL >
         {% if defined(utm_term) %} AND url LIKE concat('%utm_term=', encodeURLFormComponent({{ String(utm_term) }}), '%') {% end %}
         {% if defined(utm_content) %} AND url LIKE concat('%utm_content=', encodeURLFormComponent({{ String(utm_content) }}), '%') {% end %}
         {% if defined(url) %} AND url = {{ url }} {% end %}
+
+        {% if defined(filters) %}
+            {% for item in JSON(filters, '[]') %}
+                {% if item.get('operator', '') == 'equals' %}
+                    AND JSONExtractString(metadata, {{ item.get('operand', '') }}) = {{ item.get('value', '') }}
+                {% end %}
+            {% end %}
+        {% end %}
+
     ORDER BY timestamp {% if order == 'asc' %} ASC {% else %} DESC {% end %}
     LIMIT {{ Int32(limit, 100) }}
     {% if defined(offset) %} OFFSET {{ Int32(offset, 0) }} {% end %}