fix: improvements in the dashboard

packages/db/src/clickhouse/query-builder.ts

@@ -247,11 +247,16 @@ export class Query<T = any> {
   }
 
   // Fill
-  fill(from: string | Date, to: string | Date, step: string): this {
+  fill(
+    from: string | Date | Expression,
+    to: string | Date | Expression,
+    step: string | Expression,
+  ): this {
     this._fill = {
-      from: this.escapeDate(from),
-      to: this.escapeDate(to),
-      step: step,
+      from:
+        from instanceof Expression ? from.toString() : this.escapeDate(from),
+      to: to instanceof Expression ? to.toString() : this.escapeDate(to),
+      step: step instanceof Expression ? step.toString() : step,
     };
     return this;
   }

packages/db/src/services/overview.service.ts

@@ -165,16 +165,6 @@ export class OverviewService {
       views_per_session: number;
     }[];
   }> {
-    console.log('-----------------');
-    console.log('getMetrics', {
-      projectId,
-      filters,
-      startDate,
-      endDate,
-      interval,
-      timezone,
-    });
-
     const where = this.getRawWhereClause('sessions', filters);
     if (this.isPageFilter(filters)) {
       // Session aggregation with bounce rates

packages/trpc/src/routers/overview.ts

@@ -1,4 +1,7 @@
 import {
+  TABLE_NAMES,
+  ch,
+  clix,
   eventBuffer,
   getChartPrevStartEndDate,
   getChartStartEndDate,
@@ -14,8 +17,12 @@ import { format } from 'date-fns';
 import { z } from 'zod';
 import { cacheMiddleware, createTRPCRouter, publicProcedure } from '../trpc';
 
-const cacher = cacheMiddleware((input) => {
+const cacher = cacheMiddleware((input, opts) => {
   const range = input.range as IChartRange;
+  if (opts.path === 'overview.liveData') {
+    return 0;
+  }
+
   switch (range) {
     case '30min':
     case 'today':
@@ -82,6 +89,125 @@ export const overviewRouter = createTRPCRouter({
     .query(async ({ input }) => {
       return eventBuffer.getActiveVisitorCount(input.projectId);
     }),
+
+  liveData: publicProcedure
+    .input(z.object({ projectId: z.string() }))
+    .use(cacher)
+    .query(async ({ input }) => {
+      const { timezone } = await getSettingsForProject(input.projectId);
+
+      // Get total unique sessions in the last 30 minutes
+      const totalSessionsQuery = clix(ch, timezone)
+        .select<{ total_sessions: number }>([
+          'uniq(session_id) as total_sessions',
+        ])
+        .from(TABLE_NAMES.events)
+        .where('project_id', '=', input.projectId)
+        .where('name', '=', 'session_start')
+        .where('created_at', '>=', clix.exp('now() - INTERVAL 30 MINUTE'));
+
+      // Get counts per minute for the last 30 minutes
+      const minuteCountsQuery = clix(ch, timezone)
+        .select<{
+          minute: string;
+          session_count: number;
+          visitor_count: number;
+        }>([
+          `${clix.toStartOf('created_at', 'minute')} as minute`,
+          'uniq(session_id) as session_count',
+          'uniq(profile_id) as visitor_count',
+        ])
+        .from(TABLE_NAMES.events)
+        .where('project_id', '=', input.projectId)
+        .where('name', 'IN', ['session_start', 'screen_view'])
+        .where('created_at', '>=', clix.exp('now() - INTERVAL 30 MINUTE'))
+        .groupBy(['minute'])
+        .orderBy('minute', 'ASC')
+        .fill(
+          clix.exp('now() - INTERVAL 30 MINUTE'),
+          clix.exp('now()'),
+          clix.exp('INTERVAL 1 MINUTE'),
+        );
+
+      // Get referrers per minute for the last 30 minutes
+      const minuteReferrersQuery = clix(ch, timezone)
+        .select<{
+          minute: string;
+          referrer_name: string;
+          count: number;
+        }>([
+          `${clix.toStartOf('created_at', 'minute')} as minute`,
+          'referrer_name',
+          'count(*) as count',
+        ])
+        .from(TABLE_NAMES.events)
+        .where('project_id', '=', input.projectId)
+        .where('name', '=', 'session_start')
+        .where('created_at', '>=', clix.exp('now() - INTERVAL 30 MINUTE'))
+        .where('referrer_name', '!=', '')
+        .where('referrer_name', 'IS NOT NULL')
+        .groupBy(['minute', 'referrer_name'])
+        .orderBy('minute', 'ASC')
+        .orderBy('count', 'DESC');
+
+      // Get unique referrers in the last 30 minutes
+      const referrersQuery = clix(ch, timezone)
+        .select<{ referrer: string; count: number }>([
+          'referrer_name as referrer',
+          'count(*) as count',
+        ])
+        .from(TABLE_NAMES.events)
+        .where('project_id', '=', input.projectId)
+        .where('name', '=', 'session_start')
+        .where('created_at', '>=', clix.exp('now() - INTERVAL 30 MINUTE'))
+        .where('referrer_name', '!=', '')
+        .where('referrer_name', 'IS NOT NULL')
+        .groupBy(['referrer_name'])
+        .orderBy('count', 'DESC')
+        .limit(10);
+
+      const [totalSessions, minuteCounts, minuteReferrers, referrers] =
+        await Promise.all([
+          totalSessionsQuery.execute(),
+          minuteCountsQuery.execute(),
+          minuteReferrersQuery.execute(),
+          referrersQuery.execute(),
+        ]);
+
+      // Group referrers by minute
+      const referrersByMinute = new Map<
+        string,
+        Array<{ referrer: string; count: number }>
+      >();
+      minuteReferrers.forEach((item) => {
+        if (!referrersByMinute.has(item.minute)) {
+          referrersByMinute.set(item.minute, []);
+        }
+        referrersByMinute.get(item.minute)!.push({
+          referrer: item.referrer_name,
+          count: item.count,
+        });
+      });
+
+      return {
+        totalSessions: totalSessions[0]?.total_sessions || 0,
+        minuteCounts: minuteCounts.map((item) => ({
+          minute: item.minute,
+          sessionCount: item.session_count,
+          visitorCount: item.visitor_count,
+          timestamp: new Date(item.minute).getTime(),
+          time: new Date(item.minute).toLocaleTimeString([], {
+            hour: '2-digit',
+            minute: '2-digit',
+          }),
+          referrers: referrersByMinute.get(item.minute) || [],
+        })),
+        referrers: referrers.map((item) => ({
+          referrer: item.referrer,
+          count: item.count,
+        })),
+      };
+    }),
   stats: publicProcedure
     .input(
       zGetMetricsInput.omit({ startDate: true, endDate: true }).extend({

packages/trpc/src/trpc.ts

@@ -169,8 +169,15 @@ const middlewareMarker = 'middlewareMarker' as 'middlewareMarker' & {
   __brand: 'middlewareMarker';
 };
 
-export const cacheMiddleware = (cbOrTtl: number | ((input: any) => number)) =>
+export const cacheMiddleware = (
+  cbOrTtl: number | ((input: any, opts: { path: string }) => number),
+) =>
   t.middleware(async ({ ctx, next, path, type, getRawInput, input }) => {
+    const ttl =
+      typeof cbOrTtl === 'function' ? cbOrTtl(input, { path }) : cbOrTtl;
+    if (!ttl) {
+      return next();
+    }
     const rawInput = await getRawInput();
     if (type !== 'query') {
       return next();
@@ -194,7 +201,7 @@ export const cacheMiddleware = (cbOrTtl: number | ((input: any) => number)) =>
     if (result.data) {
       getRedisCache().setJson(
         key,
-        typeof cbOrTtl === 'function' ? cbOrTtl(input) : cbOrTtl,
+        ttl,
         // @ts-expect-error
         result.data,
       );