Fixed identify. Closes #3409

This commit is contained in:
Mike Cao 2025-05-09 16:54:49 -07:00
parent d3977eef7d
commit 3f1ecf4c1b
4 changed files with 156 additions and 212 deletions

View file

@ -187,26 +187,19 @@ export async function POST(request: Request) {
websiteId, websiteId,
sessionId, sessionId,
visitId, visitId,
createdAt,
// Page
pageTitle: safeDecodeURIComponent(title),
hostname: hostname || urlDomain,
urlPath: safeDecodeURI(urlPath), urlPath: safeDecodeURI(urlPath),
urlQuery, urlQuery,
utmSource,
utmMedium,
utmCampaign,
utmContent,
utmTerm,
referrerPath: safeDecodeURI(referrerPath), referrerPath: safeDecodeURI(referrerPath),
referrerQuery, referrerQuery,
referrerDomain, referrerDomain,
pageTitle: safeDecodeURIComponent(title),
gclid, // Session
fbclid, distinctId: id,
msclkid,
ttclid,
lifatid,
twclid,
eventName: name,
eventData: data,
hostname: hostname || urlDomain,
browser, browser,
os, os,
device, device,
@ -215,17 +208,31 @@ export async function POST(request: Request) {
country, country,
region, region,
city, city,
// Events
eventName: name,
eventData: data,
tag, tag,
distinctId: id,
createdAt, // UTM
utmSource,
utmMedium,
utmCampaign,
utmContent,
utmTerm,
// Click IDs
gclid,
fbclid,
msclkid,
ttclid,
lifatid,
twclid,
}); });
} }
if (type === COLLECTION_TYPE.identify) { if (type === COLLECTION_TYPE.identify) {
if (!data) { if (data) {
return badRequest('Data required.');
}
await saveSessionData({ await saveSessionData({
websiteId, websiteId,
sessionId, sessionId,
@ -234,6 +241,7 @@ export async function POST(request: Request) {
createdAt, createdAt,
}); });
} }
}
const token = createToken({ websiteId, sessionId, visitId, iat }, secret()); const token = createToken({ websiteId, sessionId, visitId, iat }, secret());

View file

@ -6,30 +6,23 @@ import prisma from '@/lib/prisma';
import { uuid } from '@/lib/crypto'; import { uuid } from '@/lib/crypto';
import { saveEventData } from './saveEventData'; import { saveEventData } from './saveEventData';
export async function saveEvent(args: { export interface SaveEventArgs {
websiteId: string; websiteId: string;
sessionId: string; sessionId: string;
visitId: string; visitId: string;
createdAt?: Date;
// Page
pageTitle?: string;
hostname?: string;
urlPath: string; urlPath: string;
urlQuery?: string; urlQuery?: string;
utmSource?: string;
utmMedium?: string;
utmCampaign?: string;
utmContent?: string;
utmTerm?: string;
referrerPath?: string; referrerPath?: string;
referrerQuery?: string; referrerQuery?: string;
referrerDomain?: string; referrerDomain?: string;
pageTitle?: string;
gclid?: string; // Session
fbclid?: string; distinctId?: string;
msclkid?: string;
ttclid?: string;
lifatid?: string;
twclid?: string;
eventName?: string;
eventData?: any;
hostname?: string;
browser?: string; browser?: string;
os?: string; os?: string;
device?: string; device?: string;
@ -38,73 +31,65 @@ export async function saveEvent(args: {
country?: string; country?: string;
region?: string; region?: string;
city?: string; city?: string;
tag?: string;
distinctId?: string;
createdAt?: Date;
}) {
return runQuery({
[PRISMA]: () => relationalQuery(args),
[CLICKHOUSE]: () => clickhouseQuery(args),
});
}
async function relationalQuery(data: { // Events
websiteId: string; eventName?: string;
sessionId: string; eventData?: any;
visitId: string; tag?: string;
urlPath: string;
urlQuery?: string; // UTM
utmSource?: string; utmSource?: string;
utmMedium?: string; utmMedium?: string;
utmCampaign?: string; utmCampaign?: string;
utmContent?: string; utmContent?: string;
utmTerm?: string; utmTerm?: string;
referrerPath?: string;
referrerQuery?: string; // Click IDs
referrerDomain?: string;
gclid?: string; gclid?: string;
fbclid?: string; fbclid?: string;
msclkid?: string; msclkid?: string;
ttclid?: string; ttclid?: string;
lifatid?: string; lifatid?: string;
twclid?: string; twclid?: string;
pageTitle?: string; }
eventName?: string;
eventData?: any; export async function saveEvent(args: SaveEventArgs) {
tag?: string; return runQuery({
hostname?: string; [PRISMA]: () => relationalQuery(args),
createdAt?: Date; [CLICKHOUSE]: () => clickhouseQuery(args),
}) { });
const { }
async function relationalQuery({
websiteId, websiteId,
sessionId, sessionId,
visitId, visitId,
createdAt,
pageTitle,
tag,
hostname,
urlPath, urlPath,
urlQuery, urlQuery,
utmSource,
utmMedium,
utmCampaign,
utmContent,
utmTerm,
referrerPath, referrerPath,
referrerQuery, referrerQuery,
referrerDomain, referrerDomain,
eventName, eventName,
eventData, eventData,
pageTitle, utmSource,
utmMedium,
utmCampaign,
utmContent,
utmTerm,
gclid, gclid,
fbclid, fbclid,
msclkid, msclkid,
ttclid, ttclid,
lifatid, lifatid,
twclid, twclid,
tag, }: SaveEventArgs) {
hostname,
createdAt,
} = data;
const websiteEventId = uuid(); const websiteEventId = uuid();
const websiteEvent = prisma.client.websiteEvent.create({ await prisma.client.websiteEvent.create({
data: { data: {
id: websiteEventId, id: websiteEventId,
websiteId, websiteId,
@ -146,83 +131,49 @@ async function relationalQuery(data: {
createdAt, createdAt,
}); });
} }
return websiteEvent;
} }
async function clickhouseQuery(data: { async function clickhouseQuery({
websiteId: string;
sessionId: string;
visitId: string;
urlPath: string;
urlQuery?: string;
utmSource?: string;
utmMedium?: string;
utmCampaign?: string;
utmContent?: string;
utmTerm?: string;
referrerPath?: string;
referrerQuery?: string;
referrerDomain?: string;
pageTitle?: string;
gclid?: string;
fbclid?: string;
msclkid?: string;
ttclid?: string;
lifatid?: string;
twclid?: string;
eventName?: string;
eventData?: any;
hostname?: string;
browser?: string;
os?: string;
device?: string;
screen?: string;
language?: string;
country?: string;
region?: string;
city?: string;
tag?: string;
distinctId?: string;
createdAt?: Date;
}) {
const {
websiteId, websiteId,
sessionId, sessionId,
visitId, visitId,
distinctId,
createdAt,
pageTitle,
browser,
os,
device,
screen,
language,
country,
region,
city,
tag,
hostname,
urlPath, urlPath,
urlQuery, urlQuery,
referrerPath,
referrerQuery,
referrerDomain,
eventName,
eventData,
utmSource, utmSource,
utmMedium, utmMedium,
utmCampaign, utmCampaign,
utmContent, utmContent,
utmTerm, utmTerm,
referrerPath,
referrerQuery,
referrerDomain,
gclid, gclid,
fbclid, fbclid,
msclkid, msclkid,
ttclid, ttclid,
lifatid, lifatid,
twclid, twclid,
pageTitle, }: SaveEventArgs) {
eventName,
eventData,
country,
region,
city,
tag,
distinctId,
createdAt,
...args
} = data;
const { insert, getUTCString } = clickhouse; const { insert, getUTCString } = clickhouse;
const { sendMessage } = kafka; const { sendMessage } = kafka;
const eventId = uuid(); const eventId = uuid();
const message = { const message = {
...args,
website_id: websiteId, website_id: websiteId,
session_id: sessionId, session_id: sessionId,
visit_id: visitId, visit_id: visitId,
@ -252,6 +203,12 @@ async function clickhouseQuery(data: {
tag: tag, tag: tag,
distinct_id: distinctId, distinct_id: distinctId,
created_at: getUTCString(createdAt), created_at: getUTCString(createdAt),
browser,
os,
device,
screen,
language,
hostname,
}; };
if (kafka.enabled) { if (kafka.enabled) {
@ -271,6 +228,4 @@ async function clickhouseQuery(data: {
createdAt, createdAt,
}); });
} }
return data;
} }

View file

@ -1,4 +1,3 @@
import { Prisma } from '@prisma/client';
import { DATA_TYPE } from '@/lib/constants'; import { DATA_TYPE } from '@/lib/constants';
import { uuid } from '@/lib/crypto'; import { uuid } from '@/lib/crypto';
import { CLICKHOUSE, PRISMA, runQuery } from '@/lib/db'; import { CLICKHOUSE, PRISMA, runQuery } from '@/lib/db';
@ -8,7 +7,7 @@ import kafka from '@/lib/kafka';
import prisma from '@/lib/prisma'; import prisma from '@/lib/prisma';
import { DynamicData } from '@/lib/types'; import { DynamicData } from '@/lib/types';
export async function saveEventData(data: { export interface SaveEventDataArgs {
websiteId: string; websiteId: string;
eventId: string; eventId: string;
sessionId?: string; sessionId?: string;
@ -16,19 +15,16 @@ export async function saveEventData(data: {
eventName?: string; eventName?: string;
eventData: DynamicData; eventData: DynamicData;
createdAt?: Date; createdAt?: Date;
}) { }
export async function saveEventData(data: SaveEventDataArgs) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(data), [PRISMA]: () => relationalQuery(data),
[CLICKHOUSE]: () => clickhouseQuery(data), [CLICKHOUSE]: () => clickhouseQuery(data),
}); });
} }
async function relationalQuery(data: { async function relationalQuery(data: SaveEventDataArgs) {
websiteId: string;
eventId: string;
eventData: DynamicData;
createdAt?: Date;
}): Promise<Prisma.BatchPayload> {
const { websiteId, eventId, eventData, createdAt } = data; const { websiteId, eventId, eventData, createdAt } = data;
const jsonKeys = flattenJSON(eventData); const jsonKeys = flattenJSON(eventData);
@ -46,20 +42,12 @@ async function relationalQuery(data: {
createdAt, createdAt,
})); }));
return prisma.client.eventData.createMany({ await prisma.client.eventData.createMany({
data: flattenedData, data: flattenedData,
}); });
} }
async function clickhouseQuery(data: { async function clickhouseQuery(data: SaveEventDataArgs) {
websiteId: string;
eventId: string;
sessionId?: string;
urlPath?: string;
eventName?: string;
eventData: DynamicData;
createdAt?: Date;
}) {
const { websiteId, sessionId, eventId, urlPath, eventName, eventData, createdAt } = data; const { websiteId, sessionId, eventId, urlPath, eventName, eventData, createdAt } = data;
const { insert, getUTCString } = clickhouse; const { insert, getUTCString } = clickhouse;
@ -88,6 +76,4 @@ async function clickhouseQuery(data: {
} else { } else {
await insert('event_data', messages); await insert('event_data', messages);
} }
return data;
} }

View file

@ -7,28 +7,29 @@ import { CLICKHOUSE, PRISMA, runQuery } from '@/lib/db';
import kafka from '@/lib/kafka'; import kafka from '@/lib/kafka';
import clickhouse from '@/lib/clickhouse'; import clickhouse from '@/lib/clickhouse';
export async function saveSessionData(data: { export interface SaveSessionDataArgs {
websiteId: string; websiteId: string;
sessionId: string; sessionId: string;
sessionData: DynamicData; sessionData: DynamicData;
distinctId?: string; distinctId?: string;
createdAt?: Date; createdAt?: Date;
}) { }
export async function saveSessionData(data: SaveSessionDataArgs) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(data), [PRISMA]: () => relationalQuery(data),
[CLICKHOUSE]: () => clickhouseQuery(data), [CLICKHOUSE]: () => clickhouseQuery(data),
}); });
} }
export async function relationalQuery(data: { export async function relationalQuery({
websiteId: string; websiteId,
sessionId: string; sessionId,
sessionData: DynamicData; sessionData,
distinctId?: string; distinctId,
createdAt?: Date; createdAt,
}) { }: SaveSessionDataArgs) {
const { client } = prisma; const { client } = prisma;
const { websiteId, sessionId, sessionData, distinctId, createdAt } = data;
const jsonKeys = flattenJSON(sessionData); const jsonKeys = flattenJSON(sessionData);
@ -75,19 +76,15 @@ export async function relationalQuery(data: {
}); });
} }
} }
return flattenedData;
} }
async function clickhouseQuery(data: { async function clickhouseQuery({
websiteId: string; websiteId,
sessionId: string; sessionId,
sessionData: DynamicData; sessionData,
distinctId?: string; distinctId,
createdAt?: Date; createdAt,
}) { }: SaveSessionDataArgs) {
const { websiteId, sessionId, sessionData, distinctId, createdAt } = data;
const { insert, getUTCString } = clickhouse; const { insert, getUTCString } = clickhouse;
const { sendMessage } = kafka; const { sendMessage } = kafka;
@ -112,6 +109,4 @@ async function clickhouseQuery(data: {
} else { } else {
await insert('session_data', messages); await insert('session_data', messages);
} }
return data;
} }