feat: session replay

* wip

* wip

* wip

* wip

* final fixes

* comments

* fix

packages/db/code-migrations/10-add-session-replay.ts

@@ -0,0 +1,60 @@
+import fs from 'node:fs';
+import path from 'node:path';
+import { TABLE_NAMES } from '../src/clickhouse/client';
+import {
+  addColumns,
+  createTable,
+  modifyTTL,
+  runClickhouseMigrationCommands,
+} from '../src/clickhouse/migration';
+import { getIsCluster } from './helpers';
+
+export async function up() {
+  const isClustered = getIsCluster();
+
+  const sqls: string[] = [
+    ...createTable({
+      name: TABLE_NAMES.session_replay_chunks,
+      columns: [
+        '`project_id` String CODEC(ZSTD(3))',
+        '`session_id` String CODEC(ZSTD(3))',
+        '`chunk_index` UInt16',
+        '`started_at` DateTime64(3) CODEC(DoubleDelta, ZSTD(3))',
+        '`ended_at` DateTime64(3) CODEC(DoubleDelta, ZSTD(3))',
+        '`events_count` UInt16',
+        '`is_full_snapshot` Bool',
+        '`payload` String CODEC(ZSTD(6))',
+      ],
+      orderBy: ['project_id', 'session_id', 'chunk_index'],
+      partitionBy: 'toYYYYMM(started_at)',
+      settings: {
+        index_granularity: 8192,
+      },
+      distributionHash: 'cityHash64(project_id, session_id)',
+      replicatedVersion: '1',
+      isClustered,
+    }),
+    modifyTTL({
+      tableName: TABLE_NAMES.session_replay_chunks,
+      isClustered,
+      ttl: 'started_at + INTERVAL 30 DAY',
+    }),
+  ];
+
+  fs.writeFileSync(
+    path.join(__filename.replace('.ts', '.sql')),
+    sqls
+      .map((sql) =>
+        sql
+          .trim()
+          .replace(/;$/, '')
+          .replace(/\n{2,}/g, '\n')
+          .concat(';'),
+      )
+      .join('\n\n---\n\n'),
+  );
+
+  if (!process.argv.includes('--dry')) {
+    await runClickhouseMigrationCommands(sqls);
+  }
+}

packages/db/code-migrations/migrate.ts

@@ -19,12 +19,19 @@ async function migrate() {
   const migration = args.filter((arg) => !arg.startsWith('--'))[0];
 
   const migrationsDir = path.join(__dirname, '..', 'code-migrations');
-  const migrations = fs.readdirSync(migrationsDir).filter((file) => {
-    const version = file.split('-')[0];
-    return (
-      !Number.isNaN(Number.parseInt(version ?? '')) && file.endsWith('.ts')
-    );
-  });
+  const migrations = fs
+    .readdirSync(migrationsDir)
+    .filter((file) => {
+      const version = file.split('-')[0];
+      return (
+        !Number.isNaN(Number.parseInt(version ?? '')) && file.endsWith('.ts')
+      );
+    })
+    .sort((a, b) => {
+      const aVersion = Number.parseInt(a.split('-')[0]!);
+      const bVersion = Number.parseInt(b.split('-')[0]!);
+      return aVersion - bVersion;
+    });
 
   const finishedMigrations = await db.codeMigration.findMany();
 

packages/db/src/buffers/index.ts

@@ -2,6 +2,7 @@ import { BotBuffer as BotBufferRedis } from './bot-buffer';
 import { EventBuffer as EventBufferRedis } from './event-buffer';
 import { ProfileBackfillBuffer } from './profile-backfill-buffer';
 import { ProfileBuffer as ProfileBufferRedis } from './profile-buffer';
+import { ReplayBuffer } from './replay-buffer';
 import { SessionBuffer } from './session-buffer';
 
 export const eventBuffer = new EventBufferRedis();
@@ -9,5 +10,7 @@ export const profileBuffer = new ProfileBufferRedis();
 export const botBuffer = new BotBufferRedis();
 export const sessionBuffer = new SessionBuffer();
 export const profileBackfillBuffer = new ProfileBackfillBuffer();
+export const replayBuffer = new ReplayBuffer();
 
 export type { ProfileBackfillEntry } from './profile-backfill-buffer';
+export type { IClickhouseSessionReplayChunk } from './replay-buffer';

packages/db/src/buffers/replay-buffer.ts

@@ -0,0 +1,92 @@
+import { getSafeJson } from '@openpanel/json';
+import { getRedisCache } from '@openpanel/redis';
+import { TABLE_NAMES, ch } from '../clickhouse/client';
+import { BaseBuffer } from './base-buffer';
+
+export interface IClickhouseSessionReplayChunk {
+  project_id: string;
+  session_id: string;
+  chunk_index: number;
+  started_at: string;
+  ended_at: string;
+  events_count: number;
+  is_full_snapshot: boolean;
+  payload: string;
+}
+
+export class ReplayBuffer extends BaseBuffer {
+  private batchSize = process.env.REPLAY_BUFFER_BATCH_SIZE
+    ? Number.parseInt(process.env.REPLAY_BUFFER_BATCH_SIZE, 10)
+    : 500;
+  private chunkSize = process.env.REPLAY_BUFFER_CHUNK_SIZE
+    ? Number.parseInt(process.env.REPLAY_BUFFER_CHUNK_SIZE, 10)
+    : 500;
+
+  private readonly redisKey = 'replay-buffer';
+
+  constructor() {
+    super({
+      name: 'replay',
+      onFlush: async () => {
+        await this.processBuffer();
+      },
+    });
+  }
+
+  async add(chunk: IClickhouseSessionReplayChunk) {
+    try {
+      const redis = getRedisCache();
+      const result = await redis
+        .multi()
+        .rpush(this.redisKey, JSON.stringify(chunk))
+        .incr(this.bufferCounterKey)
+        .llen(this.redisKey)
+        .exec();
+
+      const bufferLength = (result?.[2]?.[1] as number) ?? 0;
+      if (bufferLength >= this.batchSize) {
+        await this.tryFlush();
+      }
+    } catch (error) {
+      this.logger.error('Failed to add replay chunk to buffer', { error });
+    }
+  }
+
+  async processBuffer() {
+    const redis = getRedisCache();
+    try {
+      const items = await redis.lrange(this.redisKey, 0, this.batchSize - 1);
+
+      if (items.length === 0) {
+        return;
+      }
+
+      const chunks = items
+        .map((item) => getSafeJson<IClickhouseSessionReplayChunk>(item))
+        .filter((item): item is IClickhouseSessionReplayChunk => item != null);
+
+      for (const chunk of this.chunks(chunks, this.chunkSize)) {
+        await ch.insert({
+          table: TABLE_NAMES.session_replay_chunks,
+          values: chunk,
+          format: 'JSONEachRow',
+        });
+      }
+
+      await redis
+        .multi()
+        .ltrim(this.redisKey, items.length, -1)
+        .decrby(this.bufferCounterKey, items.length)
+        .exec();
+
+      this.logger.debug('Processed replay chunks', { count: items.length });
+    } catch (error) {
+      this.logger.error('Failed to process replay buffer', { error });
+    }
+  }
+
+  async getBufferSize() {
+    const redis = getRedisCache();
+    return this.getBufferSizeWithCounter(() => redis.llen(this.redisKey));
+  }
+}

packages/db/src/buffers/session-buffer.ts

@@ -1,8 +1,7 @@
-import { type Redis, getRedisCache } from '@openpanel/redis';
-
 import { getSafeJson } from '@openpanel/json';
+import { getRedisCache, type Redis } from '@openpanel/redis';
 import { assocPath, clone } from 'ramda';
-import { TABLE_NAMES, ch } from '../clickhouse/client';
+import { ch, TABLE_NAMES } from '../clickhouse/client';
 import type { IClickhouseEvent } from '../services/event.service';
 import type { IClickhouseSession } from '../services/session.service';
 import { BaseBuffer } from './base-buffer';
@@ -35,14 +34,14 @@ export class SessionBuffer extends BaseBuffer {
       | {
           projectId: string;
           profileId: string;
-        },
+        }
   ) {
     let hit: string | null = null;
     if ('sessionId' in options) {
       hit = await this.redis.get(`session:${options.sessionId}`);
     } else {
       hit = await this.redis.get(
-        `session:${options.projectId}:${options.profileId}`,
+        `session:${options.projectId}:${options.profileId}`
       );
     }
 
@@ -54,7 +53,7 @@ export class SessionBuffer extends BaseBuffer {
   }
 
   async getSession(
-    event: IClickhouseEvent,
+    event: IClickhouseEvent
   ): Promise<[IClickhouseSession] | [IClickhouseSession, IClickhouseSession]> {
     const existingSession = await this.getExistingSession({
       sessionId: event.session_id,
@@ -186,14 +185,14 @@ export class SessionBuffer extends BaseBuffer {
         `session:${newSession.id}`,
         JSON.stringify(newSession),
         'EX',
-        60 * 60,
+        60 * 60
       );
       if (newSession.profile_id) {
         multi.set(
           `session:${newSession.project_id}:${newSession.profile_id}`,
           JSON.stringify(newSession),
           'EX',
-          60 * 60,
+          60 * 60
         );
       }
       for (const session of sessions) {
@@ -220,10 +219,12 @@ export class SessionBuffer extends BaseBuffer {
       const events = await this.redis.lrange(
         this.redisKey,
         0,
-        this.batchSize - 1,
+        this.batchSize - 1
       );
 
-      if (events.length === 0) return;
+      if (events.length === 0) {
+        return;
+      }
 
       const sessions = events
         .map((e) => getSafeJson<IClickhouseSession>(e))
@@ -258,7 +259,7 @@ export class SessionBuffer extends BaseBuffer {
     }
   }
 
-  async getBufferSize() {
+  getBufferSize() {
     return this.getBufferSizeWithCounter(() => this.redis.llen(this.redisKey));
   }
 }

packages/db/src/clickhouse/client.ts

@@ -59,6 +59,7 @@ export const TABLE_NAMES = {
   cohort_events_mv: 'cohort_events_mv',
   sessions: 'sessions',
   events_imports: 'events_imports',
+  session_replay_chunks: 'session_replay_chunks',
 };
 
 /**

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

@@ -1,19 +1,21 @@
+import { getSafeJson } from '@openpanel/json';
 import { cacheable } from '@openpanel/redis';
 import type { IChartEventFilter } from '@openpanel/validation';
 import sqlstring from 'sqlstring';
 import {
-  TABLE_NAMES,
   ch,
   chQuery,
+  convertClickhouseDateToJs,
   formatClickhouseDate,
+  TABLE_NAMES,
 } from '../clickhouse/client';
 import { clix } from '../clickhouse/query-builder';
 import { createSqlBuilder } from '../sql-builder';
 import { getEventFiltersWhereClause } from './chart.service';
 import { getOrganizationByProjectIdCached } from './organization.service';
-import { type IServiceProfile, getProfilesCached } from './profile.service';
+import { getProfilesCached, type IServiceProfile } from './profile.service';
 
-export type IClickhouseSession = {
+export interface IClickhouseSession {
   id: string;
   profile_id: string;
   event_count: number;
@@ -52,7 +54,9 @@ export type IClickhouseSession = {
   revenue: number;
   sign: 1 | 0;
   version: number;
-};
+  // Dynamically added
+  has_replay?: boolean;
+}
 
 export interface IServiceSession {
   id: string;
@@ -91,6 +95,7 @@ export interface IServiceSession {
   utmTerm: string;
   revenue: number;
   profile?: IServiceProfile;
+  hasReplay?: boolean;
 }
 
 export interface GetSessionListOptions {
@@ -114,8 +119,8 @@ export function transformSession(session: IClickhouseSession): IServiceSession {
     entryOrigin: session.entry_origin,
     exitPath: session.exit_path,
     exitOrigin: session.exit_origin,
-    createdAt: new Date(session.created_at),
-    endedAt: new Date(session.ended_at),
+    createdAt: convertClickhouseDateToJs(session.created_at),
+    endedAt: convertClickhouseDateToJs(session.ended_at),
     referrer: session.referrer,
     referrerName: session.referrer_name,
     referrerType: session.referrer_type,
@@ -142,19 +147,18 @@ export function transformSession(session: IClickhouseSession): IServiceSession {
     utmTerm: session.utm_term,
     revenue: session.revenue,
     profile: undefined,
+    hasReplay: session.has_replay,
   };
 }
 
-type Direction = 'initial' | 'next' | 'prev';
-
-type PageInfo = {
+interface PageInfo {
   next?: Cursor; // use last row
-};
+}
 
-type Cursor = {
+interface Cursor {
   createdAt: string; // ISO 8601 with ms
   id: string;
-};
+}
 
 export async function getSessionList({
   cursor,
@@ -176,8 +180,9 @@ export async function getSessionList({
     sb.where.range = `created_at BETWEEN toDateTime('${formatClickhouseDate(startDate)}') AND toDateTime('${formatClickhouseDate(endDate)}')`;
   }
 
-  if (profileId)
+  if (profileId) {
     sb.where.profileId = `profile_id = ${sqlstring.escape(profileId)}`;
+  }
   if (search) {
     const s = sqlstring.escape(`%${search}%`);
     sb.where.search = `(entry_path ILIKE ${s} OR exit_path ILIKE ${s} OR referrer ILIKE ${s} OR referrer_name ILIKE ${s})`;
@@ -191,13 +196,11 @@ export async function getSessionList({
   const dateIntervalInDays =
     organization?.subscriptionPeriodEventsLimit &&
     organization?.subscriptionPeriodEventsLimit > 1_000_000
-      ? 1
+      ? 2
       : 360;
 
   if (cursor) {
     const cAt = sqlstring.escape(cursor.createdAt);
-    // TODO: remove id from cursor
-    const cId = sqlstring.escape(cursor.id);
     sb.where.cursor = `created_at < toDateTime64(${cAt}, 3)`;
     sb.where.cursorWindow = `created_at >= toDateTime64(${cAt}, 3) - INTERVAL ${dateIntervalInDays} DAY`;
     sb.orderBy.created_at = 'created_at DESC';
@@ -235,10 +238,14 @@ export async function getSessionList({
     sb.select[column] = column;
   });
 
+  sb.select.has_replay = `toBool(src.session_id != '') as hasReplay`;
+  sb.joins.has_replay = `LEFT JOIN (SELECT DISTINCT session_id FROM ${TABLE_NAMES.session_replay_chunks} WHERE project_id = ${sqlstring.escape(projectId)} AND started_at > now() - INTERVAL ${dateIntervalInDays} DAY) AS src ON src.session_id = id`;
+
   const sql = getSql();
   const data = await chQuery<
     IClickhouseSession & {
       latestCreatedAt: string;
+      hasReplay: boolean;
     }
   >(sql);
 
@@ -321,23 +328,79 @@ export async function getSessionsCount({
 
 export const getSessionsCountCached = cacheable(getSessionsCount, 60 * 10);
 
+export interface ISessionReplayChunkMeta {
+  chunk_index: number;
+  started_at: string;
+  ended_at: string;
+  events_count: number;
+  is_full_snapshot: boolean;
+}
+
+const REPLAY_CHUNKS_PAGE_SIZE = 40;
+
+export async function getSessionReplayChunksFrom(
+  sessionId: string,
+  projectId: string,
+  fromIndex: number
+) {
+  const rows = await chQuery<{ chunk_index: number; payload: string }>(
+    `SELECT chunk_index, payload
+     FROM ${TABLE_NAMES.session_replay_chunks}
+     WHERE session_id = ${sqlstring.escape(sessionId)}
+       AND project_id = ${sqlstring.escape(projectId)}
+     ORDER BY started_at, ended_at, chunk_index
+     LIMIT ${REPLAY_CHUNKS_PAGE_SIZE + 1}
+     OFFSET ${fromIndex}`
+  );
+
+  return {
+    data: rows
+      .slice(0, REPLAY_CHUNKS_PAGE_SIZE)
+      .map((row, index) => {
+        const events = getSafeJson<
+          { type: number; data: unknown; timestamp: number }[]
+        >(row.payload);
+        if (!events) {
+          return null;
+        }
+        return { chunkIndex: index + fromIndex, events };
+      })
+      .filter(Boolean),
+    hasMore: rows.length > REPLAY_CHUNKS_PAGE_SIZE,
+  };
+}
+
 class SessionService {
   constructor(private client: typeof ch) {}
 
   async byId(sessionId: string, projectId: string) {
-    const result = await clix(this.client)
-      .select<IClickhouseSession>(['*'])
-      .from(TABLE_NAMES.sessions)
-      .where('id', '=', sessionId)
-      .where('project_id', '=', projectId)
-      .where('sign', '=', 1)
-      .execute();
+    const [sessionRows, hasReplayRows] = await Promise.all([
+      clix(this.client)
+        .select<IClickhouseSession>(['*'])
+        .from(TABLE_NAMES.sessions, true)
+        .where('id', '=', sessionId)
+        .where('project_id', '=', projectId)
+        .where('sign', '=', 1)
+        .execute(),
+      chQuery<{ n: number }>(
+        `SELECT 1 AS n
+         FROM ${TABLE_NAMES.session_replay_chunks}
+         WHERE session_id = ${sqlstring.escape(sessionId)}
+           AND project_id = ${sqlstring.escape(projectId)}
+         LIMIT 1`
+      ),
+    ]);
 
-    if (!result[0]) {
+    if (!sessionRows[0]) {
       throw new Error('Session not found');
     }
 
-    return transformSession(result[0]);
+    const session = transformSession(sessionRows[0]);
+
+    return {
+      ...session,
+      hasReplay: hasReplayRows.length > 0,
+    };
   }
 }