Tinybird Analytics

Copyimport { Tinybird } from "@chronark/zod-bird";
 
export const tb = new Tinybird({
  token: process.env.TINYBIRD_API_KEY as string,
  baseUrl: process.env.TINYBIRD_API_URL as string,
});

Copyexport const recordClickZod = tb.buildIngestEndpoint({
  datasource: "dub_click_events",
  event: recordClickZodSchema,
  wait: true,
});

Copyexport async function recordClick({
  req,
  clickId,
  workspaceId,
  linkId,
  domain,
  key,
  url,
  programId,
  partnerId,
  skipRatelimit,
  timestamp,
  referrer,
  trigger = "link",
  shouldCacheClickId,
}: {
  req: Request;
  clickId?: string;
  linkId: string;
  workspaceId?: string;
  domain: string;
  key: string;
  url?: string;
  programId?: string;
  partnerId?: string;
  skipRatelimit?: boolean;
  timestamp?: string;
  referrer?: string;
  trigger?: string;
  shouldCacheClickId?: boolean;
}) {
  if (!clickId) {
    return null;
  }
 
  const searchParams = new URL(req.url).searchParams;
 
  if (req.headers.has("dub-no-track") || searchParams.has("dub-no-track")) {
    return null;
  }
 
  const ua = userAgent(req);
 
  if (trigger !== "deeplink") {
    const isBot = detectBot(req);
 
    if (isBot) {
      console.log(`Click not recorded ❌ – Bot detected.`, {
        ua,
        isBot,
      });
      return null;
    }
  }

Copy  const identityHash = await getIdentityHash(req);
 
  if (!skipRatelimit) {
    try {
      const cachedClickId = await recordClickCache.get({
        domain,
        key,
        identityHash,
      });
      if (cachedClickId) {
        return null;
      }
    } catch (error) {
      console.error(`[recordClickCache error]: ${error}`);
      return null;
    }
  }
 
  const isQr = detectQr(req);
  if (isQr) {
    trigger = "qr";
  }
 
  const { continent, region } =
    process.env.VERCEL === "1"
      ? {
          continent: req.headers.get("x-vercel-ip-continent"),
          region: req.headers.get("x-vercel-ip-country-region"),
        }
      : LOCALHOST_GEO_DATA;
 
  const geo =
    process.env.VERCEL === "1" ? geolocation(req) : LOCALHOST_GEO_DATA;
 
  const ip = process.env.VERCEL === "1" ? ipAddress(req) : LOCALHOST_IP;
  const isEuCountry = geo.country && EU_COUNTRY_CODES.includes(geo.country);

Copy        fetchWithRetry(
          `${process.env.TINYBIRD_API_URL}/v0/events?name=dub_click_events&wait=true`,
          {
            method: "POST",
            headers: {
              Authorization: `Bearer ${process.env.TINYBIRD_API_KEY}`,
            },
            body: JSON.stringify(clickData),
          },
        ).then((res) => res.json()),
 
        recordClickCache.set({
          domain,
          key,
          identityHash,
          clickId,
        }),
 
        publishLinkClickEvent({
          linkId,
          timestamp: clickData.timestamp,
          ...(workspaceId && url && { workspaceId }),
          ...(programId && partnerId && { programId, partnerId }),
        }),
 
        publishWorkspaceClickEvent(clickData),

Copyexport const workspaceClickEventStream = new RedisStream(STREAM_KEY);
 
export const publishWorkspaceClickEvent = async (event) => {
  try {
    const parsedEvent = clickEventSchemaTB.parse({
      ...event,
      qr: Number(event.qr),
      bot: Number(event.bot),
    });
 
    const hasLinkClickedWebhooks = await clickWebhookWorkspaces.has(
      parsedEvent.workspace_id,
    );
 
    if (!hasLinkClickedWebhooks) {
      return;
    }
 
    return await redis.xadd(STREAM_KEY, "*", parsedEvent);
  } catch (error) {
    logger.error("stream.publish_failed", {
      service: "upstash",
      streamKey: STREAM_KEY,
      error: toErrorFields(error),
      correlation: {
        workspaceId: event.workspace_id,
        clickId: event.click_id,
        linkId: event.link_id,
      },
    });
 
    await logger.flush();
    throw error;
  }
};

Copyconst transformLinkTB = (link: ExpandedLink) => {
  const key = decodeKeyIfCaseSensitive({
    domain: link.domain,
    key: link.key,
  });
 
  return {
    link_id: link.id,
    domain: link.domain,
    key,
    url: link.url,
    tag_ids: link.tags?.map(({ tag }) => tag.id) ?? [],
    folder_id: link.folderId ?? "",
    tenant_id: link.tenantId ?? "",
    program_id: link.programId ?? "",
    partner_id: link.partnerId ?? "",
    partner_group_id: link.programEnrollment?.groupId ?? "",
    partner_tag_ids:
      link.programEnrollment?.programPartnerTags?.map(
        ({ partnerTagId }) => partnerTagId,
      ) ?? [],
    workspace_id: link.projectId,
    created_at: link.createdAt,
  };
};

Copyexport const recordLink = async (
  payload: ExpandedLink | ExpandedLink[],
  { deleted }: { deleted?: boolean } = {},
) => {
  if (Array.isArray(payload)) {
    return await recordLinkTB(
      payload.map(transformLinkTB).map((p) => ({ ...p, deleted })),
    );
  } else {
    return await recordLinkTB({ ...transformLinkTB(payload), deleted });
  }
};

Copy  const deleteCondition = "domain = 'xyz'";
 
  //  delete data from tinybird
  const deleteRes = await Promise.allSettled([
    deleteData({
      dataSource: "dub_links_metadata",
      deleteCondition,
    }),
    deleteData({
      dataSource: "dub_links_metadata_latest",
      deleteCondition,
    }),
    deleteData({
      dataSource: "dub_regular_links_metadata_latest",
      deleteCondition,
    }),
  ]);
  console.log(deleteRes);
}

Copyconst deleteData = async ({
  dataSource,
  deleteCondition,
}: {
  dataSource: string;
  deleteCondition: string;
}) => {
  return fetch(
    `https://api.us-east.tinybird.co/v0/datasources/${dataSource}/delete`,
    {
      method: "POST",
      headers: {
        Authorization: `Bearer ${process.env.TINYBIRD_API_KEY}`,
        "Content-Type": "application/x-www-form-urlencoded",
      },
      body: `delete_condition=${deleteCondition}`,
    },
  ).then((res) => res.json());
};

Copy  const pipe = tb.buildPipe({
    pipe: ["count", "timeseries"].includes(groupBy!)
      ? `v4_${groupBy}`
      : [
            "top_folders",
            "top_link_tags",
            "top_domains",
            "top_partners",
            "top_partner_tags",
            "top_groups",
          ].includes(groupBy!)
        ? "v4_group_by_link_metadata"
        : "v4_group_by",
    parameters: analyticsFilterTB,
    data: z.object({
      groupByField: z.string(),
      clicks: z.number().default(0),
      leads: z.number().default(0),
      sales: z.number().default(0),
      saleAmount: z.number().default(0),
      // only for cities and regions groupBy
      country: z.string().optional(),
      region: z.string().optional(),
    }),
  });

Copyconst getTopLinksByCountriesTB = tb.buildPipe({
  pipe: "v3_group_by_link_country",
  parameters: inputSchema,
  data: responseSchema,
});

Copy  const tinybirdParams: any = {
    workspaceId,
    customerId: params.customerId,
    programId: params.programId,
    partnerId: partnerIdParam,
    partnerIdOperator,
    partnerTagId: partnerTagIdParam,
    partnerTagIdOperator,
    tenantId: tenantIdParam,
    tenantIdOperator,
    groupId: groupIdParam,
    groupIdOperator,
    linkId: linkIdParam,
    linkIdOperator,
    domain: domainParam,
    domainOperator,
    folderId: folderIdParam,
    folderIdOperator,
    tagId: tagIdParam,
    tagIdOperator,
    groupBy,
    eventType: event,
    start: formatUTCDateTimeClickhouse(startDate),
    end: formatUTCDateTimeClickhouse(endDate),
    granularity,
    timezone,
    region: typeof regionForPipe === "string" ? regionForPipe : undefined,
    root: params.root,
    saleType: params.saleType,
    filters: allFilters.length > 0 ? JSON.stringify(allFilters) : undefined,
  };

Copyconst payloads = await Promise.all(
  data.map(async ({ projectId, _count, _sum }) => {
    const [topLinks, topCountries] = (await Promise.all([
      getAnalytics({
        workspaceId: projectId!,
        event: "clicks",
        groupBy: "top_links",
        interval: "ytd",
        root: false,
      }),
      getAnalytics({
        workspaceId: projectId!,
        event: "clicks",
        groupBy: "countries",
        interval: "ytd",
      }),
    ])) as [AnalyticsTopLinks[], AnalyticsCountries[]];
 
    return {
      year: 2024,
      workspaceId: projectId!,
      totalLinks: _count.id,
      totalClicks: _sum.clicks ?? 0,
      topLinks: topLinks.slice(0, 5).map(({ domain, key, clicks }) => ({
        item: `${domain}/${key}`,
        count: clicks,
      })),
      topCountries: topCountries.slice(0, 5).map(({ country, clicks }) => ({
        item: country,
        count: clicks,
      })),
    };
  }),
);

Copyconst getClickEventTB = tb.buildPipe({
  pipe: "get_click_event",
  parameters: z.object({
    clickId: z.string(),
  }),
  data: clickEventSchemaTB,
});

Copyexport const getCustomerEventsTB = async ({
  customerId,
  linkIds,
  limit,
}: {
  customerId: string;
  linkIds?: string[];
  limit?: number;
}) => {
  return await pipe({
    customerId,
    ...(linkIds ? { linkIds } : {}),
    ...(limit ? { limit } : {}),
  });
};

Copy  const pipe = tb.buildPipe({
    pipe: "v4_events",
    parameters: eventsFilterTB,
    data:
      {
        clicks: clickEventSchemaTBEndpoint,
        leads: leadEventSchemaTBEndpoint,
        sales: saleEventSchemaTBEndpoint,
      }[eventType] ?? clickEventSchemaTBEndpoint,
  });

Copyconst deleteData = async ({
  dataSource,
  clickId,
  columnName,
  oldValue,
}: {
  dataSource: string;
  clickId: string;
  columnName: string;
  oldValue: string;
}) => {
  return fetch(
    `https://api.us-east.tinybird.co/v0/datasources/${dataSource}/delete`,
    {
      method: "POST",
      headers: {
        Authorization: `Bearer ${process.env.TINYBIRD_API_KEY}`,
        "Content-Type": "application/x-www-form-urlencoded",
      },
      body: `delete_condition=click_id='${clickId}' and ${columnName}='${oldValue}'`,
    },
  ).then((res) => res.json());
};

Copyconst aggregateClickStats = (
  entries: RedisStreamEntry<LinkClickEvent>[],
): {
  linkUpdates: LinkAggregate[];
  workspaceUpdates: WorkspaceAggregate[];
  enrollmentUpdates: EnrollmentAggregate[];
} => {
  const links = new Map<string, LinkAggregate>();
  const workspaces = new Map<string, WorkspaceAggregate>();
  const enrollments = new Map<string, EnrollmentAggregate>();
 
  for (const entry of entries) {
    const { linkId, workspaceId, programId, partnerId, timestamp } = entry.data;
 
    if (!linkId) {
      continue;
    }
 
    const parsedTimestamp = Date.parse(timestamp);
    const lastClicked = Number.isFinite(parsedTimestamp)
      ? parsedTimestamp
      : Date.now();
 
    const existingLink = links.get(linkId);
    if (existingLink) {
      existingLink.clicks += 1;
      existingLink.lastClicked = Math.max(
        existingLink.lastClicked,
        lastClicked,
      );
      existingLink.entryIds.push(entry.id);
    } else {
      links.set(linkId, {
        linkId,
        clicks: 1,
        lastClicked,
        entryIds: [entry.id],
      });
    }
 
    if (workspaceId) {
      const existingWorkspace = workspaces.get(workspaceId);
      if (existingWorkspace) {
        existingWorkspace.clicks += 1;
      } else {
        workspaces.set(workspaceId, {
          workspaceId,
          clicks: 1,
        });
      }
    }
 
    if (programId && partnerId) {
      const key = `${programId}:${partnerId}`;
      const existingEnrollment = enrollments.get(key);
      if (existingEnrollment) {
        existingEnrollment.clicks += 1;
      } else {
        enrollments.set(key, {
          programId,
          partnerId,
          clicks: 1,
        });
      }
    }
  }
 
  return {
    linkUpdates: Array.from(links.values()),
    workspaceUpdates: Array.from(workspaces.values()),
    enrollmentUpdates: Array.from(enrollments.values()),
  };
};

Copyconst processInSubBatches = async <T>(
  items: T[],
  handler: (item: T) => Promise<{ success: boolean; error?: unknown }>,
) => {
  const errors: unknown[] = [];
  let totalProcessed = 0;
 
  for (let i = 0; i < items.length; i += SUB_BATCH_SIZE) {
    const batch = items.slice(i, i + SUB_BATCH_SIZE);
    const results = await Promise.allSettled(batch.map(handler));
 
    for (const result of results) {
      if (result.status === "fulfilled" && result.value.success) {
        totalProcessed++;
      } else if (result.status === "fulfilled" && result.value.error) {
        errors.push(result.value.error);
      } else if (result.status === "rejected") {
        errors.push(result.reason);
      }
    }
  }
 
  return { totalProcessed, errors };
};