session activity identify stitching

db/clickhouse/migrations/13_add_session_link.sql

@@ -0,0 +1,12 @@
+-- Create session_link
+CREATE TABLE umami.session_link
+(
+    website_id UUID,
+    session_id UUID,
+    distinct_id String,
+    created_at DateTime('UTC'),
+    INDEX idx_session_id session_id TYPE bloom_filter GRANULARITY 1
+)
+ENGINE = ReplacingMergeTree
+    ORDER BY (website_id, distinct_id, session_id)
+    SETTINGS index_granularity = 8192;

db/clickhouse/schema.sql

@@ -424,3 +424,16 @@ ENGINE = MergeTree
     PARTITION BY toYYYYMM(created_at)
     ORDER BY (website_id, url_path, event_type, created_at)
     SETTINGS index_granularity = 8192;
+
+-- Create session_link
+CREATE TABLE umami.session_link
+(
+    website_id UUID,
+    session_id UUID,
+    distinct_id String,
+    created_at DateTime('UTC'),
+    INDEX idx_session_id session_id TYPE bloom_filter GRANULARITY 1
+)
+ENGINE = ReplacingMergeTree
+    ORDER BY (website_id, distinct_id, session_id)
+    SETTINGS index_granularity = 8192;

prisma/migrations/21_add_session_link/migration.sql

@@ -0,0 +1,11 @@
+-- CreateTable
+CREATE TABLE "session_link" (
+    "website_id" UUID NOT NULL,
+    "session_id" UUID NOT NULL,
+    "distinct_id" VARCHAR(50) NOT NULL,
+    "created_at" TIMESTAMPTZ(6) DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "session_link_pkey" PRIMARY KEY ("website_id","distinct_id","session_id")
+);
+
+CREATE INDEX "session_link_website_id_session_id_idx" ON "session_link"("website_id", "session_id");

prisma/schema.prisma

@@ -63,6 +63,17 @@ model Session {
   @@map("session")
 }
 
+model SessionLink {
+  websiteId  String    @map("website_id") @db.Uuid
+  sessionId  String    @map("session_id") @db.Uuid
+  distinctId String    @map("distinct_id") @db.VarChar(50)
+  createdAt  DateTime? @default(now()) @map("created_at") @db.Timestamptz(6)
+
+  @@id([websiteId, distinctId, sessionId])
+  @@index([websiteId, sessionId])
+  @@map("session_link")
+}
+
 model Website {
   id        String    @id() @map("website_id") @db.Uuid
   name      String    @db.VarChar(100)

src/app/(main)/websites/[websiteId]/sessions/SessionActivity.tsx

@@ -22,11 +22,13 @@ export function SessionActivity({
   sessionId,
   startDate,
   endDate,
+  distinctId,
 }: {
   websiteId: string;
   sessionId: string;
   startDate: Date;
   endDate: Date;
+  distinctId?: string;
 }) {
   const { t, labels } = useMessages();
   const { formatTimezoneDate } = useTimezone();
@@ -35,6 +37,7 @@ export function SessionActivity({
     sessionId,
     startDate,
     endDate,
+    distinctId,
   );
   const { isMobile } = useMobile();
   let lastDay = null;

src/app/(main)/websites/[websiteId]/sessions/SessionProfile.tsx

@@ -81,6 +81,7 @@ export function SessionProfile({
                   sessionId={sessionId}
                   startDate={data?.firstAt}
                   endDate={data?.lastAt}
+                  distinctId={data?.distinctId}
                 />
               </TabPanel>
               <TabPanel id="properties">

src/app/api/send/route.ts

@@ -11,13 +11,14 @@ import { parseRequest } from '@/lib/request';
 import { badRequest, forbidden, json, serverError } from '@/lib/response';
 import { anyObjectParam, urlOrPathParam } from '@/lib/schema';
 import { safeDecodeURI, safeDecodeURIComponent } from '@/lib/url';
-import { createSession, saveEvent, saveSessionData } from '@/queries/sql';
+import { createSession, saveEvent, saveSessionData, saveSessionLink } from '@/queries/sql';
 
 interface Cache {
   websiteId: string;
   sessionId: string;
   visitId: string;
   iat: number;
+  sessionLinkId?: string;
 }
 
 // Reject strings whose first character is a spreadsheet formula trigger to
@@ -128,6 +129,9 @@ export async function POST(request: Request) {
       }
     }
 
+    // Carried forward in the cache token so repeat identify calls skip the link write
+    let sessionLinkId = cache?.sessionLinkId;
+
     // Client info
     const { ip, userAgent, device, browser, os, country, region, city } = await getClientInfo(
       request,
@@ -277,6 +281,34 @@ export async function POST(request: Request) {
         twclid,
       });
     } else if (type === COLLECTION_TYPE.identify) {
+      if (websiteId && id) {
+        // Persist both memberships so stitched activity can be resolved from
+        // either the anonymous session or the identified session.
+        const anonymousSessionId = uuid(sourceId, ip, userAgent, sessionSalt);
+        const linkedSessionIds = [...new Set([anonymousSessionId, sessionId])];
+        const newLinkId = hash(...linkedSessionIds, id);
+
+        if (sessionLinkId !== newLinkId) {
+          // Best-effort: a link failure must not block the session data write below
+          try {
+            await Promise.all(
+              linkedSessionIds.map(linkedSessionId =>
+                saveSessionLink({
+                  websiteId,
+                  sessionId: linkedSessionId,
+                  distinctId: id,
+                  createdAt,
+                }),
+              ),
+            );
+            sessionLinkId = newLinkId;
+          } catch (e) {
+            // eslint-disable-next-line no-console
+            console.error('Failed to save session link:', e);
+          }
+        }
+      }
+
       if (data) {
         await saveSessionData({
           websiteId,
@@ -316,7 +348,7 @@ export async function POST(request: Request) {
     }
 
     const token = createToken(
-      { websiteId, sessionId, visitId, iat, type: CACHE_TOKEN_TYPE },
+      { websiteId, sessionId, visitId, iat, sessionLinkId, type: CACHE_TOKEN_TYPE },
       secret(),
     );
 

src/app/api/websites/[websiteId]/sessions/[sessionId]/activity/route.ts

@@ -1,8 +1,15 @@
 import { z } from 'zod';
+import { FIELD_LENGTH } from '@/lib/constants';
+import { uuid } from '@/lib/crypto';
 import { getQueryFilters, parseRequest } from '@/lib/request';
 import { json, unauthorized } from '@/lib/response';
 import { canViewWebsiteSection } from '@/permissions';
-import { getSessionActivity } from '@/queries/sql';
+import {
+  getLinkedDistinctIds,
+  getLinkedSessionIds,
+  getSessionActivity,
+  getSessionActivityBounds,
+} from '@/queries/sql';
 
 export async function GET(
   request: Request,
@@ -11,6 +18,7 @@ export async function GET(
   const schema = z.object({
     startAt: z.coerce.number().int(),
     endAt: z.coerce.number().int(),
+    distinctId: z.string().max(FIELD_LENGTH.distinctId).optional(),
   });
 
   const { auth, query, error } = await parseRequest(request, schema);
@@ -22,19 +30,38 @@ export async function GET(
   const { websiteId, sessionId } = await params;
 
   if (
-    !(await canViewWebsiteSection(auth, websiteId, [
-      'sessions',
-      'events',
-      'realtime',
-      'revenue',
-    ]))
+    !(await canViewWebsiteSection(auth, websiteId, ['sessions', 'events', 'realtime', 'revenue']))
   ) {
     return unauthorized();
   }
 
-  const filters = await getQueryFilters(query, websiteId);
+  let sessionIds = [sessionId];
+  let startAt = query.startAt;
+  let endAt = query.endAt;
+  const distinctIds = query.distinctId
+    ? [query.distinctId]
+    : await getLinkedDistinctIds(websiteId, sessionId);
 
-  const data = await getSessionActivity(websiteId, sessionId, filters);
+  if (distinctIds.length) {
+    const links = await Promise.all(distinctIds.map(distinctId => getLinkedSessionIds(websiteId, distinctId)));
+    const linkedIds = links.flatMap(group => group.map(link => link.sessionId));
+    const identifiedIds = distinctIds.map(distinctId => uuid(websiteId, distinctId));
+
+    sessionIds = Array.from(new Set([sessionId, ...identifiedIds, ...linkedIds]));
+
+    if (sessionIds.length > 1) {
+      const bounds = await getSessionActivityBounds(websiteId, sessionIds);
+
+      if (bounds?.firstAt && bounds?.lastAt) {
+        startAt = Math.min(startAt, +new Date(bounds.firstAt));
+        endAt = Math.max(endAt, +new Date(bounds.lastAt));
+      }
+    }
+  }
+
+  const filters = await getQueryFilters({ ...query, startAt, endAt }, websiteId);
+
+  const data = await getSessionActivity(websiteId, sessionIds, filters);
 
   return json(data);
 }

src/components/hooks/queries/useSessionActivityQuery.ts

@@ -5,15 +5,17 @@ export function useSessionActivityQuery(
   sessionId: string,
   startDate: Date,
   endDate: Date,
+  distinctId?: string,
 ) {
   const { get, useQuery } = useApi();
 
   return useQuery({
-    queryKey: ['session:activity', { websiteId, sessionId, startDate, endDate }],
+    queryKey: ['session:activity', { websiteId, sessionId, startDate, endDate, distinctId }],
     queryFn: () => {
       return get(`/websites/${websiteId}/sessions/${sessionId}/activity`, {
         startAt: +new Date(startDate),
         endAt: +new Date(endDate),
+        ...(distinctId && { distinctId }),
       });
     },
     enabled: Boolean(websiteId && sessionId && startDate && endDate),

src/queries/sql/index.ts

@@ -43,7 +43,10 @@ export * from './reports/getJourney';
 export * from './reports/getRetention';
 export * from './reports/getUTM';
 export * from './sessions/createSession';
+export * from './sessions/getLinkedDistinctIds';
+export * from './sessions/getLinkedSessionIds';
 export * from './sessions/getSessionActivity';
+export * from './sessions/getSessionActivityBounds';
 export * from './sessions/getSessionData';
 export * from './sessions/getSessionDataActivityStats';
 export * from './sessions/getSessionDataArraySeries';
@@ -61,3 +64,4 @@ export * from './sessions/getWebsiteSession';
 export * from './sessions/getWebsiteSessionStats';
 export * from './sessions/getWebsiteSessions';
 export * from './sessions/saveSessionData';
+export * from './sessions/saveSessionLink';

src/queries/sql/sessions/getLinkedDistinctIds.ts

@@ -0,0 +1,44 @@
+import clickhouse from '@/lib/clickhouse';
+import { CLICKHOUSE, PRISMA, runQuery } from '@/lib/db';
+import prisma from '@/lib/prisma';
+
+const FUNCTION_NAME = 'getLinkedDistinctIds';
+
+export async function getLinkedDistinctIds(
+  ...args: [websiteId: string, sessionId: string]
+): Promise<string[]> {
+  return runQuery({
+    [PRISMA]: () => relationalQuery(...args),
+    [CLICKHOUSE]: () => clickhouseQuery(...args),
+  });
+}
+
+async function relationalQuery(websiteId: string, sessionId: string): Promise<string[]> {
+  const { rawQuery } = prisma;
+
+  return rawQuery(
+    `
+    select distinct distinct_id as "distinctId"
+    from session_link
+    where website_id = {{websiteId::uuid}}
+      and session_id = {{sessionId::uuid}}
+    `,
+    { websiteId, sessionId },
+    FUNCTION_NAME,
+  ).then(result => (result as { distinctId: string }[]).map(({ distinctId }) => distinctId));
+}
+
+async function clickhouseQuery(websiteId: string, sessionId: string): Promise<string[]> {
+  const { rawQuery } = clickhouse;
+
+  return rawQuery(
+    `
+    select distinct distinct_id as distinctId
+    from session_link
+    where website_id = {websiteId:UUID}
+      and session_id = {sessionId:UUID}
+    `,
+    { websiteId, sessionId },
+    FUNCTION_NAME,
+  ).then(result => (result as { distinctId: string }[]).map(({ distinctId }) => distinctId));
+}

src/queries/sql/sessions/getLinkedSessionIds.ts

@@ -0,0 +1,50 @@
+import clickhouse from '@/lib/clickhouse';
+import { CLICKHOUSE, PRISMA, runQuery } from '@/lib/db';
+import prisma from '@/lib/prisma';
+
+const FUNCTION_NAME = 'getLinkedSessionIds';
+
+export async function getLinkedSessionIds(
+  ...args: [websiteId: string, distinctId: string]
+): Promise<{ sessionId: string; createdAt: string }[]> {
+  return runQuery({
+    [PRISMA]: () => relationalQuery(...args),
+    [CLICKHOUSE]: () => clickhouseQuery(...args),
+  });
+}
+
+async function relationalQuery(websiteId: string, distinctId: string) {
+  const { rawQuery } = prisma;
+
+  return rawQuery(
+    `
+    select
+      session_id as "sessionId",
+      min(created_at) as "createdAt"
+    from session_link
+    where website_id = {{websiteId::uuid}}
+      and distinct_id = {{distinctId}}
+    group by session_id
+    `,
+    { websiteId, distinctId },
+    FUNCTION_NAME,
+  );
+}
+
+async function clickhouseQuery(websiteId: string, distinctId: string) {
+  const { rawQuery } = clickhouse;
+
+  return rawQuery(
+    `
+    select
+      session_id as sessionId,
+      min(created_at) as createdAt
+    from session_link
+    where website_id = {websiteId:UUID}
+      and distinct_id = {distinctId:String}
+    group by session_id
+    `,
+    { websiteId, distinctId },
+    FUNCTION_NAME,
+  );
+}

src/queries/sql/sessions/getSessionActivity.ts

@@ -7,7 +7,7 @@ import type { QueryFilters } from '@/lib/types';
 const FUNCTION_NAME = 'getSessionActivity';
 
 export async function getSessionActivity(
-  ...args: [websiteId: string, sessionId: string, filters: QueryFilters]
+  ...args: [websiteId: string, sessionIds: string[], filters: QueryFilters]
 ) {
   return runQuery({
     [PRISMA]: () => relationalQuery(...args),
@@ -15,7 +15,7 @@ export async function getSessionActivity(
   });
 }
 
-async function relationalQuery(websiteId: string, sessionId: string, filters: QueryFilters) {
+async function relationalQuery(websiteId: string, sessionIds: string[], filters: QueryFilters) {
   const { rawQuery } = prisma;
   const { startDate, endDate } = filters;
 
@@ -37,18 +37,18 @@ async function relationalQuery(websiteId: string, sessionId: string, filters: Qu
                       and created_at between {{startDate}} and {{endDate}}) AS "hasData"
     from website_event
     where website_id = {{websiteId::uuid}}
-      and session_id = {{sessionId::uuid}}
+      and session_id = any({{sessionIds}}::uuid[])
       and event_type != ${EVENT_TYPE.performance}
       and created_at between {{startDate}} and {{endDate}}
     order by created_at desc
     limit 500
     `,
-    { websiteId, sessionId, startDate, endDate },
+    { websiteId, sessionIds, startDate, endDate },
     FUNCTION_NAME,
   );
 }
 
-async function clickhouseQuery(websiteId: string, sessionId: string, filters: QueryFilters) {
+async function clickhouseQuery(websiteId: string, sessionIds: string[], filters: QueryFilters) {
   const { rawQuery } = clickhouse;
   const { startDate, endDate } = filters;
 
@@ -64,20 +64,20 @@ async function clickhouseQuery(websiteId: string, sessionId: string, filters: Qu
       event_name as eventName,
       visit_id as visitId,
       hostname,
-      event_id IN (select event_id 
-                   from event_data 
-                   where website_id = {websiteId:UUID} 
-                    and session_id = {sessionId:UUID}
+      event_id IN (select event_id
+                   from event_data
+                   where website_id = {websiteId:UUID}
+                    and session_id in {sessionIds:Array(UUID)}
                     and created_at between {startDate:DateTime64} and {endDate:DateTime64}) AS hasData
     from website_event
     where website_id = {websiteId:UUID}
-      and session_id = {sessionId:UUID} 
+      and session_id in {sessionIds:Array(UUID)}
       and event_type != ${EVENT_TYPE.performance}
       and created_at between {startDate:DateTime64} and {endDate:DateTime64}
     order by created_at desc
     limit 500
     `,
-    { websiteId, sessionId, startDate, endDate },
+    { websiteId, sessionIds, startDate, endDate },
     FUNCTION_NAME,
   );
 }

src/queries/sql/sessions/getSessionActivityBounds.ts

@@ -0,0 +1,62 @@
+import clickhouse from '@/lib/clickhouse';
+import { EVENT_TYPE } from '@/lib/constants';
+import { CLICKHOUSE, PRISMA, runQuery } from '@/lib/db';
+import prisma from '@/lib/prisma';
+
+const FUNCTION_NAME = 'getSessionActivityBounds';
+
+interface SessionActivityBounds {
+  firstAt: string | Date;
+  lastAt: string | Date;
+}
+
+export async function getSessionActivityBounds(
+  ...args: [websiteId: string, sessionIds: string[]]
+): Promise<SessionActivityBounds | null> {
+  return runQuery({
+    [PRISMA]: () => relationalQuery(...args),
+    [CLICKHOUSE]: () => clickhouseQuery(...args),
+  });
+}
+
+async function relationalQuery(
+  websiteId: string,
+  sessionIds: string[],
+): Promise<SessionActivityBounds | null> {
+  const { rawQuery } = prisma;
+
+  return rawQuery(
+    `
+    select
+      min(created_at) as "firstAt",
+      max(created_at) as "lastAt"
+    from website_event
+    where website_id = {{websiteId::uuid}}
+      and session_id = any({{sessionIds}}::uuid[])
+      and event_type != ${EVENT_TYPE.performance}
+    `,
+    { websiteId, sessionIds },
+    FUNCTION_NAME,
+  ).then(result => (result?.[0]?.firstAt && result?.[0]?.lastAt ? result[0] : null));
+}
+
+async function clickhouseQuery(
+  websiteId: string,
+  sessionIds: string[],
+): Promise<SessionActivityBounds | null> {
+  const { rawQuery, getDateStringSQL } = clickhouse;
+
+  return rawQuery(
+    `
+    select
+      ${getDateStringSQL('min(created_at)')} as firstAt,
+      ${getDateStringSQL('max(created_at)')} as lastAt
+    from website_event
+    where website_id = {websiteId:UUID}
+      and session_id in {sessionIds:Array(UUID)}
+      and event_type != ${EVENT_TYPE.performance}
+    `,
+    { websiteId, sessionIds },
+    FUNCTION_NAME,
+  ).then(result => (result?.[0]?.firstAt && result?.[0]?.lastAt ? result[0] : null));
+}

src/queries/sql/sessions/saveSessionLink.ts

@@ -0,0 +1,79 @@
+import clickhouse from '@/lib/clickhouse';
+import { FIELD_LENGTH } from '@/lib/constants';
+import { CLICKHOUSE, PRISMA, runQuery } from '@/lib/db';
+import { truncateString } from '@/lib/format';
+import kafka from '@/lib/kafka';
+import prisma from '@/lib/prisma';
+
+const FUNCTION_NAME = 'saveSessionLink';
+
+export interface SaveSessionLinkArgs {
+  websiteId: string;
+  sessionId: string;
+  distinctId: string;
+  createdAt?: Date;
+}
+
+export async function saveSessionLink(data: SaveSessionLinkArgs) {
+  return runQuery({
+    [PRISMA]: () => relationalQuery(data),
+    [CLICKHOUSE]: () => clickhouseQuery(data),
+  });
+}
+
+async function relationalQuery({
+  websiteId,
+  sessionId,
+  distinctId,
+  createdAt,
+}: SaveSessionLinkArgs) {
+  const { writeRawQuery } = prisma;
+
+  await writeRawQuery(
+    `
+    insert into session_link (
+      website_id,
+      session_id,
+      distinct_id,
+      created_at
+    )
+    values (
+      {{websiteId}},
+      {{sessionId}},
+      {{distinctId}},
+      {{createdAt}}
+    )
+    on conflict (website_id, distinct_id, session_id) do nothing
+    `,
+    {
+      websiteId,
+      sessionId,
+      distinctId: truncateString(distinctId, FIELD_LENGTH.distinctId),
+      createdAt,
+    },
+    FUNCTION_NAME,
+  );
+}
+
+async function clickhouseQuery({
+  websiteId,
+  sessionId,
+  distinctId,
+  createdAt,
+}: SaveSessionLinkArgs) {
+  const { insert, getUTCString } = clickhouse;
+  const { sendMessage } = kafka;
+
+  const message = {
+    website_id: websiteId,
+    session_id: sessionId,
+    distinct_id: truncateString(distinctId, FIELD_LENGTH.distinctId),
+    created_at: getUTCString(createdAt),
+  };
+
+  if (kafka.enabled) {
+    await sendMessage('session_link', [message]);
+  } else {
+    await insert('session_link', [message]);
+  }
+}