session replay updates. add clickhouse implementation, filters, tables

src/queries/sql/events/getEventStats.ts

@@ -1,5 +1,4 @@
 import clickhouse from '@/lib/clickhouse';
-import { EVENT_TYPE } from '@/lib/constants';
 import { CLICKHOUSE, PRISMA, runQuery } from '@/lib/db';
 import prisma from '@/lib/prisma';
 import type { QueryFilters } from '@/lib/types';
@@ -36,7 +35,6 @@ async function relationalQuery(
   const { filterQuery, cohortQuery, joinSessionQuery, queryParams } = parseFilters({
     ...filters,
     websiteId,
-    eventType: EVENT_TYPE.customEvent,
   });
 
   const limitQuery = limit
@@ -63,6 +61,7 @@ async function relationalQuery(
     ${joinSessionQuery}
     where website_event.website_id = {{websiteId::uuid}}
       and website_event.created_at between {{startDate}} and {{endDate}}
+      and website_event.event_type = 2
       ${filterQuery}
       ${limitQuery}
     group by 1, 2
@@ -84,7 +83,6 @@ async function clickhouseQuery(
   const { filterQuery, cohortQuery, queryParams } = parseFilters({
     ...filters,
     websiteId,
-    eventType: EVENT_TYPE.customEvent,
   });
 
   const limitQuery = limit
@@ -93,7 +91,7 @@ async function clickhouseQuery(
     from website_event
     where website_id = {websiteId:UUID}
       and created_at between {startDate:DateTime64} and {endDate:DateTime64}
-      and event_type = {eventType:UInt32}
+      and event_type = 2
     group by event_name
     order by count(*) desc
     limit ${limit}
@@ -112,6 +110,7 @@ async function clickhouseQuery(
     ${cohortQuery}
     where website_id = {websiteId:UUID}
       and created_at between {startDate:DateTime64} and {endDate:DateTime64}
+      and event_type = 2
       ${filterQuery}
       ${limitQuery}
     group by x, t
@@ -129,7 +128,7 @@ async function clickhouseQuery(
       from website_event_stats_hourly website_event
       where website_id = {websiteId:UUID}
         and created_at between {startDate:DateTime64} and {endDate:DateTime64}
-        and event_type = {eventType:UInt32}
+        and event_type = 2
         ${limitQuery}
     ) as g
     group by x, t

src/queries/sql/index.ts

@@ -25,10 +25,9 @@ export * from './pageviews/getPageviewMetrics';
 export * from './pageviews/getPageviewStats';
 export * from './performance/getPerformanceStats';
 export * from './performance/savePerformance';
-export * from './replays/deleteReplaysByWebsite';
 export * from './replays/getReplayChunks';
 export * from './replays/getSessionReplays';
-export * from './replays/saveReplayChunk';
+export * from './replays/saveRecording';
 export * from './reports/getBreakdown';
 export * from './reports/getFunnel';
 export * from './reports/getJourney';

src/queries/sql/replays/deleteReplaysByWebsite.ts

@@ -1,11 +0,0 @@
-import prisma from '@/lib/prisma';
-
-export async function deleteReplaysByWebsite(websiteId: string) {
-  return relationalQuery(websiteId);
-}
-
-async function relationalQuery(websiteId: string) {
-  return prisma.client.sessionReplay.deleteMany({
-    where: { websiteId },
-  });
-}

src/queries/sql/replays/getReplayChunks.ts

@@ -1,24 +1,93 @@
+import { gunzipSync } from 'node:zlib';
+import clickhouse from '@/lib/clickhouse';
+import { CLICKHOUSE, PRISMA, runQuery } from '@/lib/db';
 import prisma from '@/lib/prisma';
 
-export async function getReplayChunks(websiteId: string, sessionId: string) {
-  return relationalQuery(websiteId, sessionId);
+const FUNCTION_NAME = 'getReplayChunks';
+
+export interface ReplayChunk {
+  events: any[];
+  chunkIndex: number;
+  eventCount: number;
+  startedAt: Date;
+  endedAt: Date;
 }
 
-async function relationalQuery(websiteId: string, sessionId: string) {
-  return prisma.client.sessionReplay.findMany({
-    where: {
-      websiteId,
-      sessionId,
-    },
-    orderBy: {
-      chunkIndex: 'asc',
-    },
-    select: {
-      events: true,
-      chunkIndex: true,
-      eventCount: true,
-      startedAt: true,
-      endedAt: true,
-    },
+export async function getReplayChunks(
+  websiteId: string,
+  sessionId: string,
+): Promise<ReplayChunk[]> {
+  return runQuery({
+    [PRISMA]: () => relationalQuery(websiteId, sessionId),
+    [CLICKHOUSE]: () => clickhouseQuery(websiteId, sessionId),
   });
 }
+
+async function relationalQuery(websiteId: string, sessionId: string): Promise<ReplayChunk[]> {
+  const { rawQuery } = prisma;
+
+  const chunks: {
+    events: Buffer;
+    chunkIndex: number;
+    eventCount: number;
+    startedAt: Date;
+    endedAt: Date;
+  }[] = await rawQuery(
+    `
+    select
+      events,
+      chunk_index as "chunkIndex",
+      event_count as "eventCount",
+      started_at as "startedAt",
+      ended_at as "endedAt"
+    from session_replay
+    where website_id = {{websiteId::uuid}}
+      and session_id = {{sessionId::uuid}}
+    order by chunk_index asc
+    `,
+    { websiteId, sessionId },
+    FUNCTION_NAME,
+  );
+
+  return chunks.map(chunk => ({
+    ...chunk,
+    events: JSON.parse(gunzipSync(Buffer.from(chunk.events)).toString('utf-8')),
+  }));
+}
+
+async function clickhouseQuery(websiteId: string, sessionId: string): Promise<ReplayChunk[]> {
+  const { rawQuery } = clickhouse;
+
+  const results = await rawQuery<
+    {
+      events: string;
+      chunk_index: number;
+      event_count: number;
+      started_at: string;
+      ended_at: string;
+    }[]
+  >(
+    `
+    select
+      events,
+      chunk_index,
+      event_count,
+      started_at,
+      ended_at
+    from session_replay
+    where website_id = {websiteId:UUID}
+      and session_id = {sessionId:UUID}
+    order by chunk_index asc
+    `,
+    { websiteId, sessionId },
+    FUNCTION_NAME,
+  );
+
+  return results.map(row => ({
+    events: JSON.parse(row.events),
+    chunkIndex: row.chunk_index,
+    eventCount: row.event_count,
+    startedAt: new Date(row.started_at),
+    endedAt: new Date(row.ended_at),
+  }));
+}

src/queries/sql/replays/getSessionReplays.ts

@@ -1,32 +1,42 @@
+import clickhouse from '@/lib/clickhouse';
+import { CLICKHOUSE, PRISMA, runQuery } from '@/lib/db';
 import prisma from '@/lib/prisma';
 import type { QueryFilters } from '@/lib/types';
 
 const FUNCTION_NAME = 'getSessionReplays';
 
-export async function getSessionReplays(...args: [websiteId: string, filters: QueryFilters]) {
-  return relationalQuery(...args);
+export function getSessionReplays(...args: [websiteId: string, filters: QueryFilters]) {
+  return runQuery({
+    [PRISMA]: () => relationalQuery(...args),
+    [CLICKHOUSE]: () => clickhouseQuery(...args),
+  });
 }
 
 async function relationalQuery(websiteId: string, filters: QueryFilters) {
   const { pagedRawQuery, parseFilters } = prisma;
-  const { search, startDate, endDate } = filters;
-  const { queryParams } = parseFilters({
+  const { search } = filters;
+  const { filterQuery, cohortQuery, queryParams } = parseFilters({
     ...filters,
     websiteId,
     search: search ? `%${search}%` : undefined,
   });
 
-  let dateQuery = '';
-  if (startDate && endDate) {
-    dateQuery = `and sr.created_at between {{startDate}} and {{endDate}}`;
-  } else if (startDate) {
-    dateQuery = `and sr.created_at >= {{startDate}}`;
-  }
+  const joinQuery =
+    filterQuery || cohortQuery
+      ? `join (select *
+               from website_event
+               where website_id = {{websiteId::uuid}}
+                  and created_at between {{startDate}} and {{endDate}}) website_event
+        on website_event.website_id = sr.website_id
+          and website_event.session_id = sr.session_id`
+      : '';
 
   const searchQuery = search
     ? `and (session.distinct_id ilike {{search}}
            or session.city ilike {{search}}
-           or session.browser ilike {{search}})`
+           or session.browser ilike {{search}}
+           or session.os ilike {{search}}
+           or session.device ilike {{search}})`
     : '';
 
   return pagedRawQuery(
@@ -46,10 +56,13 @@ async function relationalQuery(websiteId: string, filters: QueryFilters) {
       (extract(epoch from max(sr.ended_at) - min(sr.started_at)) * 1000)::bigint as "duration",
       max(sr.created_at) as "createdAt"
     from session_replay sr
-    left join session on session.session_id = sr.session_id
+    join session on session.session_id = sr.session_id
       and session.website_id = sr.website_id
+    ${cohortQuery}
+    ${joinQuery}
     where sr.website_id = {{websiteId::uuid}}
-    ${dateQuery}
+      and sr.created_at between {{startDate}} and {{endDate}}
+    ${filterQuery}
     ${searchQuery}
     group by sr.session_id,
       sr.website_id,
@@ -65,3 +78,58 @@ async function relationalQuery(websiteId: string, filters: QueryFilters) {
     FUNCTION_NAME,
   );
 }
+
+async function clickhouseQuery(websiteId: string, filters: QueryFilters) {
+  const { pagedRawQuery, parseFilters } = clickhouse;
+  const { search } = filters;
+  const { queryParams, cohortQuery, filterQuery } = parseFilters({
+    ...filters,
+    websiteId,
+  });
+
+  const searchQuery = search
+    ? `and ((positionCaseInsensitive(distinct_id, {search:String}) > 0)
+           or (positionCaseInsensitive(city, {search:String}) > 0)
+           or (positionCaseInsensitive(browser, {search:String}) > 0)
+           or (positionCaseInsensitive(os, {search:String}) > 0)
+           or (positionCaseInsensitive(device, {search:String}) > 0))`
+    : '';
+
+  return pagedRawQuery(
+    `
+    select
+      session_replay.session_id as id,
+      session_replay.website_id as websiteId,
+      website_event.browser,
+      website_event.os,
+      website_event.device,
+      website_event.country,
+      website_event.city,
+      sum(session_replay.event_count) as eventCount,
+      count(session_replay.replay_id) as chunkCount,
+      min(session_replay.started_at) as startedAt,
+      max(session_replay.ended_at) as endedAt,
+      toInt64(dateDiff('millisecond', min(session_replay.started_at), max(session_replay.ended_at))) as duration,
+      max(session_replay.created_at) as createdAt
+    from session_replay
+    join (
+      select *
+      from website_event
+      where website_id = {websiteId:UUID}
+        and created_at between {startDate:DateTime64} and {endDate:DateTime64}
+    ) website_event
+    on website_event.session_id = session_replay.session_id
+      and website_event.website_id = session_replay.website_id
+    ${cohortQuery}
+    where session_replay.website_id = {websiteId:UUID}
+        and session_replay.created_at between {startDate:DateTime64} and {endDate:DateTime64}
+    ${filterQuery}
+    ${searchQuery}
+    group by session_replay.session_id, session_replay.website_id, website_event.browser, website_event.os, website_event.device, website_event.country, website_event.city
+    order by max(created_at) desc
+    `,
+    queryParams,
+    filters,
+    FUNCTION_NAME,
+  );
+}

src/queries/sql/replays/index.ts

@@ -1,4 +0,0 @@
-export * from './deleteReplaysByWebsite';
-export * from './getReplayChunks';
-export * from './getSessionReplays';
-export * from './saveReplayChunk';

src/queries/sql/replays/saveRecording.ts

@@ -0,0 +1,72 @@
+import { gzipSync } from 'node:zlib';
+import clickhouse from '@/lib/clickhouse';
+import { uuid } from '@/lib/crypto';
+import { CLICKHOUSE, PRISMA, runQuery } from '@/lib/db';
+import prisma from '@/lib/prisma';
+
+export interface SaveRecordingArgs {
+  websiteId: string;
+  sessionId: string;
+  chunkIndex: number;
+  events: any[];
+  eventCount: number;
+  startedAt: Date;
+  endedAt: Date;
+}
+
+export async function saveRecording(args: SaveRecordingArgs) {
+  return runQuery({
+    [PRISMA]: () => relationalQuery(args),
+    [CLICKHOUSE]: () => clickhouseQuery(args),
+  });
+}
+
+async function relationalQuery({
+  websiteId,
+  sessionId,
+  chunkIndex,
+  events,
+  eventCount,
+  startedAt,
+  endedAt,
+}: SaveRecordingArgs) {
+  const compressed = gzipSync(Buffer.from(JSON.stringify(events), 'utf-8'));
+
+  return prisma.client.sessionReplay.create({
+    data: {
+      id: uuid(),
+      websiteId,
+      sessionId,
+      chunkIndex,
+      events: compressed as any,
+      eventCount,
+      startedAt,
+      endedAt,
+    },
+  });
+}
+
+async function clickhouseQuery({
+  websiteId,
+  sessionId,
+  chunkIndex,
+  events,
+  eventCount,
+  startedAt,
+  endedAt,
+}: SaveRecordingArgs) {
+  const { insert, getUTCString } = clickhouse;
+
+  return insert('session_replay', [
+    {
+      replay_id: uuid(),
+      website_id: websiteId,
+      session_id: sessionId,
+      chunk_index: chunkIndex,
+      events: JSON.stringify(events),
+      event_count: eventCount,
+      started_at: getUTCString(startedAt),
+      ended_at: getUTCString(endedAt),
+    },
+  ]);
+}

src/queries/sql/replays/saveReplayChunk.ts

@@ -1,39 +0,0 @@
-import { uuid } from '@/lib/crypto';
-import prisma from '@/lib/prisma';
-
-export interface SaveReplayChunkArgs {
-  websiteId: string;
-  sessionId: string;
-  chunkIndex: number;
-  events: Uint8Array;
-  eventCount: number;
-  startedAt: Date;
-  endedAt: Date;
-}
-
-export async function saveReplayChunk(args: SaveReplayChunkArgs) {
-  return relationalQuery(args);
-}
-
-async function relationalQuery({
-  websiteId,
-  sessionId,
-  chunkIndex,
-  events,
-  eventCount,
-  startedAt,
-  endedAt,
-}: SaveReplayChunkArgs) {
-  return prisma.client.sessionReplay.create({
-    data: {
-      id: uuid(),
-      websiteId,
-      sessionId,
-      chunkIndex,
-      events: new Uint8Array(events) as any,
-      eventCount,
-      startedAt,
-      endedAt,
-    },
-  });
-}

src/queries/sql/sessions/getWebsiteSessionStats.ts

@@ -70,7 +70,7 @@ async function clickhouseQuery(
       uniq(session_id) as "visitors",
       uniq(visit_id) as "visits",
       uniq(country) as "countries",
-      sum(length(event_name)) as "events"
+      sumIf(1, event_type = 2) as "events"
     from website_event
     ${cohortQuery}
     where website_id = {websiteId:UUID}

src/queries/sql/sessions/getWebsiteSessions.ts

@@ -48,6 +48,7 @@ async function relationalQuery(websiteId: string, filters: QueryFilters) {
       max(website_event.created_at) as "lastAt",
       count(distinct website_event.visit_id) as "visits",
       sum(case when website_event.event_type = 1 then 1 else 0 end) as "views",
+      sum(case when website_event.event_type = 2 then 1 else 0 end) as "events",
       max(website_event.created_at) as "createdAt"
     from website_event 
     ${cohortQuery}
@@ -112,6 +113,7 @@ async function clickhouseQuery(websiteId: string, filters: QueryFilters) {
       ${getDateStringSQL('max(created_at)')} as lastAt,
       uniq(visit_id) as visits,
       sumIf(1, event_type = 1) as views,
+      sumIf(1, event_type = 2) as events,
       lastAt as createdAt
     from website_event
     ${cohortQuery}
@@ -140,6 +142,7 @@ async function clickhouseQuery(websiteId: string, filters: QueryFilters) {
       ${getDateStringSQL('max(max_time)')} as lastAt,
       uniq(visit_id) as visits,
       sumIf(views, event_type = 1) as views,
+      sum(length(event_name)) as events,
       lastAt as createdAt
     from website_event_stats_hourly as website_event
     ${cohortQuery}