diff --git a/db/clickhouse/migrations/13_add_session_link.sql b/db/clickhouse/migrations/13_add_session_link.sql new file mode 100644 index 000000000..2455c05e7 --- /dev/null +++ b/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; diff --git a/db/clickhouse/schema.sql b/db/clickhouse/schema.sql index abcbe1434..1030538c1 100644 --- a/db/clickhouse/schema.sql +++ b/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; diff --git a/prisma/migrations/21_add_session_link/migration.sql b/prisma/migrations/21_add_session_link/migration.sql new file mode 100644 index 000000000..1be077332 --- /dev/null +++ b/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"); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 1c285de47..2a7c49e99 100644 --- a/prisma/schema.prisma +++ b/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) diff --git a/src/app/(main)/websites/[websiteId]/sessions/SessionActivity.tsx b/src/app/(main)/websites/[websiteId]/sessions/SessionActivity.tsx index 1a54f506a..78a62815d 100644 --- a/src/app/(main)/websites/[websiteId]/sessions/SessionActivity.tsx +++ b/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; diff --git a/src/app/(main)/websites/[websiteId]/sessions/SessionProfile.tsx b/src/app/(main)/websites/[websiteId]/sessions/SessionProfile.tsx index ffc6ce7f1..1c0405472 100644 --- a/src/app/(main)/websites/[websiteId]/sessions/SessionProfile.tsx +++ b/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} /> diff --git a/src/app/api/send/route.ts b/src/app/api/send/route.ts index 6d2086176..e07ba2971 100644 --- a/src/app/api/send/route.ts +++ b/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(), ); diff --git a/src/app/api/websites/[websiteId]/sessions/[sessionId]/activity/route.ts b/src/app/api/websites/[websiteId]/sessions/[sessionId]/activity/route.ts index 23eb78ab8..5af7660b7 100644 --- a/src/app/api/websites/[websiteId]/sessions/[sessionId]/activity/route.ts +++ b/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); } diff --git a/src/components/hooks/queries/useSessionActivityQuery.ts b/src/components/hooks/queries/useSessionActivityQuery.ts index d8d34aca6..08f1a3b87 100644 --- a/src/components/hooks/queries/useSessionActivityQuery.ts +++ b/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), diff --git a/src/queries/sql/index.ts b/src/queries/sql/index.ts index a4be400d0..2139e58af 100644 --- a/src/queries/sql/index.ts +++ b/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'; diff --git a/src/queries/sql/sessions/getLinkedDistinctIds.ts b/src/queries/sql/sessions/getLinkedDistinctIds.ts new file mode 100644 index 000000000..d3d9953cb --- /dev/null +++ b/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 { + return runQuery({ + [PRISMA]: () => relationalQuery(...args), + [CLICKHOUSE]: () => clickhouseQuery(...args), + }); +} + +async function relationalQuery(websiteId: string, sessionId: string): Promise { + 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 { + 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)); +} diff --git a/src/queries/sql/sessions/getLinkedSessionIds.ts b/src/queries/sql/sessions/getLinkedSessionIds.ts new file mode 100644 index 000000000..58f67651f --- /dev/null +++ b/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, + ); +} diff --git a/src/queries/sql/sessions/getSessionActivity.ts b/src/queries/sql/sessions/getSessionActivity.ts index f8cff7f39..c71b251b9 100644 --- a/src/queries/sql/sessions/getSessionActivity.ts +++ b/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, ); } diff --git a/src/queries/sql/sessions/getSessionActivityBounds.ts b/src/queries/sql/sessions/getSessionActivityBounds.ts new file mode 100644 index 000000000..02debf8eb --- /dev/null +++ b/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 { + return runQuery({ + [PRISMA]: () => relationalQuery(...args), + [CLICKHOUSE]: () => clickhouseQuery(...args), + }); +} + +async function relationalQuery( + websiteId: string, + sessionIds: string[], +): Promise { + 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 { + 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)); +} diff --git a/src/queries/sql/sessions/saveSessionLink.ts b/src/queries/sql/sessions/saveSessionLink.ts new file mode 100644 index 000000000..7fcabe000 --- /dev/null +++ b/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]); + } +}