feat: write recording summary events

.run/Plugin Server.run.xml

@@ -10,9 +10,11 @@
       <env name="CLICKHOUSE_SECURE" value="False" />
       <env name="DATABASE_URL" value="postgres://posthog:posthog@localhost:5432/posthog" />
       <env name="KAFKA_HOSTS" value="localhost:9092" />
-      <env name="WORKER_CONCURRENCY" value="2" />
       <env name="OBJECT_STORAGE_ENABLED" value="True" />
+      <env name="SESSION_RECORDING_SUMMARY_INGESTION_ENABLED_TEAMS" value="all" />
+      <env name="WORKER_CONCURRENCY" value="2" />
+      <env name="SESSION_RECORDING_BLOB_PROCESSING_TEAMS" value="all" />
     </envs>
     <method v="2" />
   </configuration>
-</component>
+</component>

plugin-server/src/config/config.ts

@@ -113,13 +113,15 @@ export function getDefaultConfig(): PluginsServerConfig {
         USE_KAFKA_FOR_SCHEDULED_TASKS: true,
         CLOUD_DEPLOYMENT: 'default', // Used as a Sentry tag
 
-        SESSION_RECORDING_BLOB_PROCESSING_TEAMS: '', // TODO: BW Change this to 'all' when we release it fully
+        SESSION_RECORDING_BLOB_PROCESSING_TEAMS: '', // TODO: Change this to 'all' when we release it fully
         SESSION_RECORDING_LOCAL_DIRECTORY: '.tmp/sessions',
         SESSION_RECORDING_MAX_BUFFER_AGE_SECONDS: 60 * 10, // NOTE: 10 minutes
         SESSION_RECORDING_MAX_BUFFER_SIZE_KB: ['dev', 'test'].includes(process.env.NODE_ENV || 'undefined')
             ? 1024 // NOTE: ~100KB in dev or test, so that even with gzipped content we still flush pretty frequently
             : 1024 * 50, // ~50MB after compression in prod
         SESSION_RECORDING_REMOTE_FOLDER: 'session_recordings',
+
+        SESSION_RECORDING_SUMMARY_INGESTION_ENABLED_TEAMS: '', // TODO: Change this to 'all' when we release it fully
     }
 }
 

plugin-server/src/config/kafka-topics.ts

@@ -10,6 +10,7 @@ export const KAFKA_PERSON = `${prefix}clickhouse_person${suffix}`
 export const KAFKA_PERSON_UNIQUE_ID = `${prefix}clickhouse_person_unique_id${suffix}`
 export const KAFKA_PERSON_DISTINCT_ID = `${prefix}clickhouse_person_distinct_id${suffix}`
 export const KAFKA_CLICKHOUSE_SESSION_RECORDING_EVENTS = `${prefix}clickhouse_session_recording_events${suffix}`
+export const KAFKA_CLICKHOUSE_SESSION_REPLAY_EVENTS = `${prefix}clickhouse_session_replay_events${suffix}`
 export const KAFKA_PERFORMANCE_EVENTS = `${prefix}clickhouse_performance_events${suffix}`
 export const KAFKA_EVENTS_PLUGIN_INGESTION = `${prefix}events_plugin_ingestion${suffix}`
 export const KAFKA_EVENTS_PLUGIN_INGESTION_OVERFLOW = `${prefix}events_plugin_ingestion_overflow${suffix}`

plugin-server/src/main/ingestion-queues/session-recording/session-recordings-consumer.ts

@@ -3,6 +3,7 @@ import { HighLevelProducer as RdKafkaProducer, Message, NumberNullUndefined } fr
 
 import {
     KAFKA_CLICKHOUSE_SESSION_RECORDING_EVENTS,
+    KAFKA_CLICKHOUSE_SESSION_REPLAY_EVENTS,
     KAFKA_PERFORMANCE_EVENTS,
     KAFKA_SESSION_RECORDING_EVENTS,
     KAFKA_SESSION_RECORDING_EVENTS_DLQ,
@@ -14,7 +15,11 @@ import { createKafkaProducer, disconnectProducer, flushProducer, produce } from
 import { PipelineEvent, RawEventMessage, Team } from '../../../types'
 import { KafkaConfig } from '../../../utils/db/hub'
 import { status } from '../../../utils/status'
-import { createPerformanceEvent, createSessionRecordingEvent } from '../../../worker/ingestion/process-event'
+import {
+    createPerformanceEvent,
+    createSessionRecordingEvent,
+    createSessionReplayEvent,
+} from '../../../worker/ingestion/process-event'
 import { TeamManager } from '../../../worker/ingestion/team-manager'
 import { parseEventTimestamp } from '../../../worker/ingestion/timestamps'
 import { eventDroppedCounter } from '../metrics'
@@ -25,12 +30,14 @@ export const startSessionRecordingEventsConsumer = async ({
     consumerMaxBytes,
     consumerMaxBytesPerPartition,
     consumerMaxWaitMs,
+    summaryIngestionEnabledTeams,
 }: {
     teamManager: TeamManager
     kafkaConfig: KafkaConfig
     consumerMaxBytes: number
     consumerMaxBytesPerPartition: number
     consumerMaxWaitMs: number
+    summaryIngestionEnabledTeams: string
 }) => {
     /*
         For Session Recordings we need to prepare the data for ClickHouse.
@@ -59,7 +66,13 @@ export const startSessionRecordingEventsConsumer = async ({
 
     const connectionConfig = createRdConnectionConfigFromEnvVars(kafkaConfig)
     const producer = await createKafkaProducer(connectionConfig)
-    const eachBatchWithContext = eachBatch({ teamManager, producer })
+
+    const eachBatchWithContext = eachBatch({
+        teamManager,
+        producer,
+        summaryEnabledTeams:
+            summaryIngestionEnabledTeams === 'all' ? null : summaryIngestionEnabledTeams.split(',').map(parseInt),
+    })
 
     // Create a node-rdkafka consumer that fetches batches of messages, runs
     // eachBatchWithContext, then commits offsets for the batch.
@@ -84,7 +97,15 @@ export const startSessionRecordingEventsConsumer = async ({
 }
 
 export const eachBatch =
-    ({ teamManager, producer }: { teamManager: TeamManager; producer: RdKafkaProducer }) =>
+    ({
+        teamManager,
+        producer,
+        summaryEnabledTeams,
+    }: {
+        teamManager: TeamManager
+        producer: RdKafkaProducer
+        summaryEnabledTeams: number[] | null
+    }) =>
     async (messages: Message[]) => {
         // To start with, we simply process each message in turn,
         // without attempting to perform any concurrency. There is a lot
@@ -105,7 +126,7 @@ export const eachBatch =
         // DependencyUnavailableError error to distinguish between
         // intermittent and permanent errors.
         const pendingProduceRequests: Promise<NumberNullUndefined>[] = []
-        const eachMessageWithContext = eachMessage({ teamManager, producer })
+        const eachMessageWithContext = eachMessage({ teamManager, producer, summaryEnabledTeams })
 
         for (const message of messages) {
             const results = await retryOnDependencyUnavailableError(() => eachMessageWithContext(message))
@@ -148,7 +169,15 @@ export const eachBatch =
     }
 
 const eachMessage =
-    ({ teamManager, producer }: { teamManager: TeamManager; producer: RdKafkaProducer }) =>
+    ({
+        teamManager,
+        producer,
+        summaryEnabledTeams,
+    }: {
+        teamManager: TeamManager
+        producer: RdKafkaProducer
+        summaryEnabledTeams: number[] | null
+    }) =>
     async (message: Message) => {
         // For each message, we:
         //
@@ -252,14 +281,34 @@ const eachMessage =
                         event.properties || {}
                     )
 
-                    return [
+                    const producePromises = [
                         produce(
                             producer,
                             KAFKA_CLICKHOUSE_SESSION_RECORDING_EVENTS,
                             Buffer.from(JSON.stringify(clickHouseRecord)),
                             message.key ? Buffer.from(message.key) : null
                         ),
                     ]
+
+                    if (summaryEnabledTeams === null || summaryEnabledTeams?.includes(team.id)) {
+                        const replayRecord = createSessionReplayEvent(
+                            messagePayload.uuid,
+                            team.id,
+                            messagePayload.distinct_id,
+                            parseEventTimestamp(event as PluginEvent),
+                            event.ip,
+                            event.properties || {}
+                        )
+                        producePromises.push(
+                            produce(
+                                producer,
+                                KAFKA_CLICKHOUSE_SESSION_REPLAY_EVENTS,
+                                Buffer.from(JSON.stringify(replayRecord)),
+                                message.key ? Buffer.from(message.key) : null
+                            )
+                        )
+                    }
+                    return producePromises
                 } else if (event.event === '$performance_event') {
                     const clickHouseRecord = createPerformanceEvent(
                         messagePayload.uuid,

plugin-server/src/main/pluginsServer.ts

@@ -418,6 +418,7 @@ export async function startPluginsServer(
                 consumerMaxBytes: serverConfig.KAFKA_CONSUMPTION_MAX_BYTES,
                 consumerMaxBytesPerPartition: serverConfig.KAFKA_CONSUMPTION_MAX_BYTES_PER_PARTITION,
                 consumerMaxWaitMs: serverConfig.KAFKA_CONSUMPTION_MAX_WAIT_MS,
+                summaryIngestionEnabledTeams: serverConfig.SESSION_RECORDING_SUMMARY_INGESTION_ENABLED_TEAMS,
             })
             stopSessionRecordingEventsConsumer = stop
             joinSessionRecordingEventsConsumer = join

plugin-server/src/types.ts

@@ -187,6 +187,8 @@ export interface PluginsServerConfig {
     SESSION_RECORDING_MAX_BUFFER_AGE_SECONDS: number
     SESSION_RECORDING_MAX_BUFFER_SIZE_KB: number
     SESSION_RECORDING_REMOTE_FOLDER: string
+
+    SESSION_RECORDING_SUMMARY_INGESTION_ENABLED_TEAMS: string
 }
 
 export interface Hub extends PluginsServerConfig {

plugin-server/src/worker/ingestion/process-event.ts

@@ -278,6 +278,79 @@ export const createSessionRecordingEvent = (
 
     return data
 }
+
+export interface SummarizedSessionRecordingEvent {
+    uuid: string
+    timestamp: string
+    team_id: number
+    distinct_id: string
+    session_id: string
+    window_id: string | undefined
+    created_at: string
+    url: string | undefined
+    click_count: number
+    keypress_count: number
+    mouse_activity_count: number
+}
+
+interface RRWebEventSummaryData {
+    href?: string
+    source?: number
+}
+
+interface RRWebEventSummary {
+    timestamp: number
+    type: number
+    data: RRWebEventSummaryData
+}
+
+export const createSessionReplayEvent = (
+    uuid: string,
+    team_id: number,
+    distinct_id: string,
+    timestamp: DateTime,
+    ip: string | null,
+    properties: Properties
+) => {
+    const timestampString = castTimestampOrNow(timestamp, TimestampFormat.ClickHouse)
+
+    const eventsSummaries: RRWebEventSummary[] = properties['$snapshot_data']?.['events_summary'] || []
+    let clickCount = 0
+    let keypressCount = 0
+    let mouseActivity = 0
+    let url: string | undefined = undefined
+    eventsSummaries.forEach((eventSummary: RRWebEventSummary) => {
+        if (eventSummary.type === 3) {
+            mouseActivity += 1
+            if (eventSummary.data.source === 2) {
+                clickCount += 1
+            }
+            if (eventSummary.data.source === 5) {
+                keypressCount += 1
+            }
+        }
+        if (!!eventSummary.data.href?.trim().length && url === undefined) {
+            url = eventSummary.data.href
+        }
+    })
+
+    const data: SummarizedSessionRecordingEvent = {
+        uuid,
+        team_id: team_id,
+        distinct_id: distinct_id,
+        session_id: properties['$session_id'],
+        window_id: properties['$window_id'],
+        timestamp: timestampString,
+        created_at: timestampString,
+        click_count: clickCount,
+        keypress_count: keypressCount,
+        mouse_activity_count: mouseActivity,
+        url: url,
+    }
+
+    return data
+}
+
 export function createPerformanceEvent(uuid: string, team_id: number, distinct_id: string, properties: Properties) {
     const data: Partial<RawPerformanceEvent> = {
         uuid,

plugin-server/tests/main/ingestion-queues/session-recording/session-recordings-consumer.test.ts

@@ -18,7 +18,7 @@ describe('session-recordings-consumer', () => {
     beforeEach(() => {
         postgres = new Pool({ connectionString: defaultConfig.DATABASE_URL })
         teamManager = new TeamManager(postgres, {} as any)
-        eachBachWithDependencies = eachBatch({ producer, teamManager })
+        eachBachWithDependencies = eachBatch({ producer, teamManager, summaryEnabledTeams: [] })
     })
 
     afterEach(() => {
@@ -57,4 +57,70 @@ describe('session-recordings-consumer', () => {
         // Should have send to the DLQ.
         expect(producer.produce).toHaveBeenCalledTimes(1)
     })
+
+    test('eachBatch emits to only one topic', async () => {
+        const organizationId = await createOrganization(postgres)
+        const teamId = await createTeam(postgres, organizationId)
+
+        await eachBachWithDependencies([
+            {
+                key: 'test',
+                value: JSON.stringify({ team_id: teamId, data: JSON.stringify({ event: '$snapshot' }) }),
+                timestamp: 123,
+            },
+        ])
+
+        expect(producer.produce).toHaveBeenCalledTimes(1)
+    })
+
+    test('eachBatch can emit to two topics', async () => {
+        const organizationId = await createOrganization(postgres)
+        const teamId = await createTeam(postgres, organizationId)
+
+        const eachBachWithDependencies: any = eachBatch({ producer, teamManager, summaryEnabledTeams: null })
+
+        await eachBachWithDependencies([
+            {
+                key: 'test',
+                value: JSON.stringify({ team_id: teamId, data: JSON.stringify({ event: '$snapshot' }) }),
+                timestamp: 123,
+            },
+        ])
+
+        expect(producer.produce).toHaveBeenCalledTimes(2)
+    })
+
+    test('eachBatch can emit to two topics for a specific team', async () => {
+        const organizationId = await createOrganization(postgres)
+        const teamId = await createTeam(postgres, organizationId)
+
+        const eachBachWithDependencies: any = eachBatch({ producer, teamManager, summaryEnabledTeams: [teamId] })
+
+        await eachBachWithDependencies([
+            {
+                key: 'test',
+                value: JSON.stringify({ team_id: teamId, data: JSON.stringify({ event: '$snapshot' }) }),
+                timestamp: 123,
+            },
+        ])
+
+        expect(producer.produce).toHaveBeenCalledTimes(2)
+    })
+
+    test('eachBatch can emit to only one topic when team is not summary enabled', async () => {
+        const organizationId = await createOrganization(postgres)
+        const teamId = await createTeam(postgres, organizationId)
+
+        const eachBachWithDependencies: any = eachBatch({ producer, teamManager, summaryEnabledTeams: [teamId + 1] })
+
+        await eachBachWithDependencies([
+            {
+                key: 'test',
+                value: JSON.stringify({ team_id: teamId, data: JSON.stringify({ event: '$snapshot' }) }),
+                timestamp: 123,
+            },
+        ])
+
+        expect(producer.produce).toHaveBeenCalledTimes(1)
+    })
 })