sync umami

This commit is contained in:
Viet-Tien Ngoc 2024-08-26 13:51:45 +07:00
commit cc4b21a070
600 changed files with 10884 additions and 3381 deletions

View file

@ -1,9 +0,0 @@
import prisma from 'lib/prisma';
export async function getSession(id: string) {
return prisma.client.session.findUnique({
where: {
id,
},
});
}

View file

@ -0,0 +1,59 @@
import clickhouse from 'lib/clickhouse';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import prisma from 'lib/prisma';
export async function getSessionActivity(
...args: [websiteId: string, sessionId: string, startDate: Date, endDate: Date]
) {
return runQuery({
[PRISMA]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args),
});
}
async function relationalQuery(
websiteId: string,
sessionId: string,
startDate: Date,
endDate: Date,
) {
return prisma.client.websiteEvent.findMany({
where: {
sessionId,
websiteId,
createdAt: { gte: startDate, lte: endDate },
},
take: 500,
orderBy: { createdAt: 'desc' },
});
}
async function clickhouseQuery(
websiteId: string,
sessionId: string,
startDate: Date,
endDate: Date,
) {
const { rawQuery } = clickhouse;
return rawQuery(
`
select
created_at as createdAt,
url_path as urlPath,
url_query as urlQuery,
referrer_domain as referrerDomain,
event_id as eventId,
event_type as eventType,
event_name as eventName,
visit_id as visitId
from website_event
where website_id = {websiteId:UUID}
and session_id = {sessionId:UUID}
and created_at between {startDate:DateTime64} and {endDate:DateTime64}
order by created_at desc
limit 500
`,
{ websiteId, sessionId, startDate, endDate },
);
}

View file

@ -0,0 +1,56 @@
import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse';
import { runQuery, PRISMA, CLICKHOUSE } from 'lib/db';
export async function getSessionData(...args: [websiteId: string, sessionId: string]) {
return runQuery({
[PRISMA]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args),
});
}
async function relationalQuery(websiteId: string, sessionId: string) {
const { rawQuery } = prisma;
return rawQuery(
`
select
website_id as "websiteId",
session_id as "sessionId",
data_key as "dataKey",
data_type as "dataType",
replace(string_value, '.0000', '') as "stringValue",
number_value as "numberValue",
date_value as "dateValue",
created_at as "createdAt"
from session_data
where website_id = {{websiteId::uuid}}
and session_id = {{sessionId::uuid}}
order by data_key asc
`,
{ websiteId, sessionId },
);
}
async function clickhouseQuery(websiteId: string, sessionId: string) {
const { rawQuery } = clickhouse;
return rawQuery(
`
select
website_id as websiteId,
session_id as sessionId,
data_key as dataKey,
data_type as dataType,
replace(string_value, '.0000', '') as stringValue,
number_value as numberValue,
date_value as dateValue,
created_at as createdAt
from session_data final
where website_id = {websiteId:UUID}
and session_id = {sessionId:UUID}
order by data_key asc
`,
{ websiteId, sessionId },
);
}

View file

@ -0,0 +1,65 @@
import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import { QueryFilters, WebsiteEventData } from 'lib/types';
export async function getSessionDataProperties(
...args: [websiteId: string, filters: QueryFilters & { propertyName?: string }]
): Promise<WebsiteEventData[]> {
return runQuery({
[PRISMA]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args),
});
}
async function relationalQuery(
websiteId: string,
filters: QueryFilters & { propertyName?: string },
) {
const { rawQuery, parseFilters } = prisma;
const { filterQuery, params } = await parseFilters(websiteId, filters, {
columns: { propertyName: 'data_key' },
});
return rawQuery(
`
select
data_key as "propertyName",
count(*) as "total"
from session_data
where website_id = {{websiteId::uuid}}
and created_at between {{startDate}} and {{endDate}}
${filterQuery}
group by data_key
order by 2 desc
limit 500
`,
params,
);
}
async function clickhouseQuery(
websiteId: string,
filters: QueryFilters & { propertyName?: string },
): Promise<{ propertyName: string; total: number }[]> {
const { rawQuery, parseFilters } = clickhouse;
const { filterQuery, params } = await parseFilters(websiteId, filters, {
columns: { propertyName: 'data_key' },
});
return rawQuery(
`
select
data_key as propertyName,
count(*) as total
from session_data final
where website_id = {websiteId:UUID}
and created_at between {startDate:DateTime64} and {endDate:DateTime64}
${filterQuery}
group by data_key
order by 2 desc
limit 500
`,
params,
);
}

View file

@ -0,0 +1,69 @@
import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import { QueryFilters, WebsiteEventData } from 'lib/types';
export async function getSessionDataValues(
...args: [websiteId: string, filters: QueryFilters & { propertyName?: string }]
): Promise<WebsiteEventData[]> {
return runQuery({
[PRISMA]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args),
});
}
async function relationalQuery(
websiteId: string,
filters: QueryFilters & { propertyName?: string },
) {
const { rawQuery, parseFilters, getDateSQL } = prisma;
const { filterQuery, params } = await parseFilters(websiteId, filters);
return rawQuery(
`
select
case
when data_type = 2 then replace(string_value, '.0000', '')
when data_type = 4 then ${getDateSQL('date_value', 'hour')}
else string_value
end as "value",
count(*) as "total"
from session_data
where website_id = {{websiteId::uuid}}
and created_at between {{startDate}} and {{endDate}}
and data_key = {{propertyName}}
${filterQuery}
group by value
order by 2 desc
limit 100
`,
params,
);
}
async function clickhouseQuery(
websiteId: string,
filters: QueryFilters & { propertyName?: string },
): Promise<{ propertyName: string; dataType: number; propertyValue: string; total: number }[]> {
const { rawQuery, parseFilters } = clickhouse;
const { filterQuery, params } = await parseFilters(websiteId, filters);
return rawQuery(
`
select
multiIf(data_type = 2, replaceAll(string_value, '.0000', ''),
data_type = 4, toString(date_trunc('hour', date_value)),
string_value) as "value",
count(*) as "total"
from session_data final
where website_id = {websiteId:UUID}
and created_at between {startDate:DateTime64} and {endDate:DateTime64}
and data_key = {propertyName:String}
${filterQuery}
group by value
order by 2 desc
limit 100
`,
params,
);
}

View file

@ -1,7 +1,7 @@
import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse';
import { runQuery, CLICKHOUSE, PRISMA } from 'lib/db';
import { EVENT_TYPE, FILTER_COLUMNS, SESSION_COLUMNS } from 'lib/constants';
import { EVENT_COLUMNS, EVENT_TYPE, FILTER_COLUMNS, SESSION_COLUMNS } from 'lib/constants';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import prisma from 'lib/prisma';
import { QueryFilters } from 'lib/types';
export async function getSessionMetrics(
@ -71,8 +71,10 @@ async function clickhouseQuery(
});
const includeCountry = column === 'city' || column === 'subdivision1';
return rawQuery(
`
let sql = '';
if (EVENT_COLUMNS.some(item => Object.keys(filters).includes(item))) {
sql = `
select
${column} x,
count(distinct session_id) y
@ -87,9 +89,27 @@ async function clickhouseQuery(
order by y desc
limit ${limit}
offset ${offset}
`,
params,
).then(a => {
`;
} else {
sql = `
select
${column} x,
uniq(session_id) y
${includeCountry ? ', country' : ''}
from website_event_stats_hourly website_event
where website_id = {websiteId:UUID}
and created_at between {startDate:DateTime64} and {endDate:DateTime64}
and event_type = {eventType:UInt32}
${filterQuery}
group by x
${includeCountry ? ', country' : ''}
order by y desc
limit ${limit}
offset ${offset}
`;
}
return rawQuery(sql, params).then(a => {
return Object.values(a).map(a => {
return { x: a.x, y: Number(a.y), country: a.country };
});

View file

@ -1,7 +1,7 @@
import clickhouse from 'lib/clickhouse';
import { EVENT_COLUMNS, EVENT_TYPE } from 'lib/constants';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import prisma from 'lib/prisma';
import { EVENT_TYPE } from 'lib/constants';
import { QueryFilters } from 'lib/types';
export async function getSessionStats(...args: [websiteId: string, filters: QueryFilters]) {
@ -13,7 +13,7 @@ export async function getSessionStats(...args: [websiteId: string, filters: Quer
async function relationalQuery(websiteId: string, filters: QueryFilters) {
const { timezone = 'utc', unit = 'day' } = filters;
const { getDateQuery, parseFilters, rawQuery } = prisma;
const { getDateSQL, parseFilters, rawQuery } = prisma;
const { filterQuery, joinSession, params } = await parseFilters(websiteId, {
...filters,
eventType: EVENT_TYPE.pageView,
@ -22,7 +22,7 @@ async function relationalQuery(websiteId: string, filters: QueryFilters) {
return rawQuery(
`
select
${getDateQuery('website_event.created_at', unit, timezone)} x,
${getDateSQL('website_event.created_at', unit, timezone)} x,
count(distinct website_event.session_id) y
from website_event
${joinSession}
@ -31,6 +31,7 @@ async function relationalQuery(websiteId: string, filters: QueryFilters) {
and event_type = {{eventType}}
${filterQuery}
group by 1
order by 1
`,
params,
);
@ -40,21 +41,23 @@ async function clickhouseQuery(
websiteId: string,
filters: QueryFilters,
): Promise<{ x: string; y: number }[]> {
const { timezone = 'UTC', unit = 'day' } = filters;
const { parseFilters, rawQuery, getDateStringQuery, getDateQuery } = clickhouse;
const { unit = 'day' } = filters;
const { parseFilters, rawQuery } = clickhouse;
const { filterQuery, params } = await parseFilters(websiteId, {
...filters,
eventType: EVENT_TYPE.pageView,
});
return rawQuery(
`
let sql = '';
if (EVENT_COLUMNS.some(item => Object.keys(filters).includes(item)) || unit === 'minute') {
sql = `
select
${getDateStringQuery('g.t', unit)} as x,
g.t as x,
g.y as y
from (
select
${getDateQuery('created_at', unit, timezone)} as t,
select
date_trunc('${unit}', created_at) as t,
count(distinct session_id) as y
from website_event
where website_id = {websiteId:UUID}
@ -64,11 +67,26 @@ async function clickhouseQuery(
group by t
) as g
order by t
`,
params,
).then(result => {
return Object.values(result).map((a: any) => {
return { x: a.x, y: Number(a.y) };
});
});
`;
} else {
sql = `
select
g.t as x,
g.y as y
from (
select
date_trunc('${unit}', created_at) as t,
uniq(session_id) as y
from website_event_stats_hourly website_event
where website_id = {websiteId:UUID}
and created_at between {startDate:DateTime64} and {endDate:DateTime64}
and event_type = {eventType:UInt32}
${filterQuery}
group by t
) as g
order by t
`;
}
return rawQuery(sql, params);
}

View file

@ -1,69 +0,0 @@
import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse';
import { runQuery, PRISMA, CLICKHOUSE } from 'lib/db';
import { QueryFilters } from 'lib/types';
export async function getSessions(...args: [websiteId: string, filters: QueryFilters]) {
return runQuery({
[PRISMA]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args),
});
}
async function relationalQuery(websiteId: string, filters: QueryFilters) {
const { startDate } = filters;
return prisma.client.session
.findMany({
where: {
websiteId,
createdAt: {
gte: startDate,
},
},
orderBy: {
createdAt: 'desc',
},
})
.then(a => {
return Object.values(a).map(a => {
return {
...a,
timestamp: new Date(a.createdAt).getTime() / 1000,
};
});
});
}
async function clickhouseQuery(websiteId: string, filters: QueryFilters) {
const { rawQuery } = clickhouse;
const { startDate } = filters;
return rawQuery(
`
select
session_id as id,
website_id as websiteId,
created_at as createdAt,
toUnixTimestamp(created_at) as timestamp,
hostname,
browser,
os,
device,
screen,
language,
country,
subdivision1,
subdivision2,
city
from website_event
where website_id = {websiteId:UUID}
and created_at >= {startDate:DateTime64}
order by created_at desc
`,
{
websiteId,
startDate,
},
);
}

View file

@ -0,0 +1,109 @@
import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse';
import { runQuery, PRISMA, CLICKHOUSE } from 'lib/db';
export async function getWebsiteSession(...args: [websiteId: string, sessionId: string]) {
return runQuery({
[PRISMA]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args),
});
}
async function relationalQuery(websiteId: string, sessionId: string) {
const { rawQuery, getTimestampDiffSQL } = prisma;
return rawQuery(
`
select id,
website_id as "websiteId",
hostname,
browser,
os,
device,
screen,
language,
country,
subdivision1,
city,
min(min_time) as "firstAt",
max(max_time) as "lastAt",
count(distinct visit_id) as visits,
sum(views) as views,
sum(events) as events,
sum(${getTimestampDiffSQL('min_time', 'max_time')}) as "totaltime"
from (select
session.session_id as id,
website_event.visit_id,
session.website_id,
session.hostname,
session.browser,
session.os,
session.device,
session.screen,
session.language,
session.country,
session.subdivision1,
session.city,
min(website_event.created_at) as min_time,
max(website_event.created_at) as max_time,
sum(case when website_event.event_type = 1 then 1 else 0 end) as views,
sum(case when website_event.event_type = 1 then 1 else 0 end) as events
from session
join website_event on website_event.session_id = session.session_id
where session.website_id = {{websiteId::uuid}}
and session.session_id = {{sessionId::uuid}}
group by session.session_id, visit_id, session.website_id, session.hostname, session.browser, session.os, session.device, session.screen, session.language, session.country, session.subdivision1, session.city) t
group by id, website_id, hostname, browser, os, device, screen, language, country, subdivision1, city;
`,
{ websiteId, sessionId },
).then(result => result?.[0]);
}
async function clickhouseQuery(websiteId: string, sessionId: string) {
const { rawQuery, getDateStringSQL } = clickhouse;
return rawQuery(
`
select id,
websiteId,
hostname,
browser,
os,
device,
screen,
language,
country,
subdivision1,
city,
${getDateStringSQL('min(min_time)')} as firstAt,
${getDateStringSQL('max(max_time)')} as lastAt,
uniq(visit_id) visits,
sum(views) as views,
sum(events) as events,
sum(max_time-min_time) as totaltime
from (select
session_id as id,
visit_id,
website_id as websiteId,
hostname,
browser,
os,
device,
screen,
language,
country,
subdivision1,
city,
min(min_time) as min_time,
max(max_time) as max_time,
sum(views) as views,
length(groupArrayArray(event_name)) as events
from website_event_stats_hourly
where website_id = {websiteId:UUID}
and session_id = {sessionId:UUID}
group by session_id, visit_id, website_id, hostname, browser, os, device, screen, language, country, subdivision1, city) t
group by id, websiteId, hostname, browser, os, device, screen, language, country, subdivision1, city;
`,
{ websiteId, sessionId },
).then(result => result?.[0]);
}

View file

@ -0,0 +1,72 @@
import clickhouse from 'lib/clickhouse';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import prisma from 'lib/prisma';
import { QueryFilters } from 'lib/types';
export async function getWebsiteSessionStats(
...args: [websiteId: string, filters: QueryFilters]
): Promise<
{ pageviews: number; visitors: number; visits: number; countries: number; events: number }[]
> {
return runQuery({
[PRISMA]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args),
});
}
async function relationalQuery(
websiteId: string,
filters: QueryFilters,
): Promise<
{ pageviews: number; visitors: number; visits: number; countries: number; events: number }[]
> {
const { parseFilters, rawQuery } = prisma;
const { filterQuery, params } = await parseFilters(websiteId, {
...filters,
});
return rawQuery(
`
select
count(*) as "pageviews",
count(distinct website_event.session_id) as "visitors",
count(distinct website_event.visit_id) as "visits",
count(distinct session.country) as "countries",
sum(case when website_event.event_type = 2 then 1 else 0 end) as "events"
from website_event
join session on website_event.session_id = session.session_id
where website_event.website_id = {{websiteId::uuid}}
and website_event.created_at between {{startDate}} and {{endDate}}
${filterQuery}
`,
params,
);
}
async function clickhouseQuery(
websiteId: string,
filters: QueryFilters,
): Promise<
{ pageviews: number; visitors: number; visits: number; countries: number; events: number }[]
> {
const { rawQuery, parseFilters } = clickhouse;
const { filterQuery, params } = await parseFilters(websiteId, {
...filters,
});
return rawQuery(
`
select
sum(views) as "pageviews",
uniq(session_id) as "visitors",
uniq(visit_id) as "visits",
uniq(country) as "countries",
sum(length(event_name)) as "events"
from umami.website_event_stats_hourly "website_event"
where website_id = {websiteId:UUID}
and created_at between {startDate:DateTime64} and {endDate:DateTime64}
${filterQuery}
`,
params,
);
}

View file

@ -0,0 +1,102 @@
import clickhouse from 'lib/clickhouse';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import prisma from 'lib/prisma';
import { PageParams, QueryFilters } from 'lib/types';
export async function getWebsiteSessions(
...args: [websiteId: string, filters?: QueryFilters, pageParams?: PageParams]
) {
return runQuery({
[PRISMA]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args),
});
}
async function relationalQuery(websiteId: string, filters: QueryFilters, pageParams: PageParams) {
const { pagedRawQuery, parseFilters } = prisma;
const { filterQuery, params } = await parseFilters(websiteId, {
...filters,
});
return pagedRawQuery(
`
with sessions as (
select
session.session_id as "id",
session.website_id as "websiteId",
session.hostname,
session.browser,
session.os,
session.device,
session.screen,
session.language,
session.country,
session.subdivision1,
session.city,
min(website_event.created_at) as "firstAt",
max(website_event.created_at) as "lastAt",
count(distinct website_event.visit_id) as "visits",
sum(case when website_event.event_type = 1 then 1 else 0 end) as "views",
max(website_event.created_at) as "createdAt"
from website_event
join session on session.session_id = website_event.session_id
where website_event.website_id = {{websiteId::uuid}}
and website_event.created_at between {{startDate}} and {{endDate}}
${filterQuery}
group by session.session_id,
session.website_id,
session.hostname,
session.browser,
session.os,
session.device,
session.screen,
session.language,
session.country,
session.subdivision1,
session.city
order by max(website_event.created_at) desc
limit 1000)
select * from sessions
`,
params,
pageParams,
);
}
async function clickhouseQuery(websiteId: string, filters: QueryFilters, pageParams?: PageParams) {
const { pagedQuery, parseFilters, getDateStringSQL } = clickhouse;
const { params, dateQuery, filterQuery } = await parseFilters(websiteId, filters);
return pagedQuery(
`
with sessions as (
select
session_id as id,
website_id as websiteId,
hostname,
browser,
os,
device,
screen,
language,
country,
subdivision1,
city,
${getDateStringSQL('min(min_time)')} as firstAt,
${getDateStringSQL('max(max_time)')} as lastAt,
uniq(visit_id) as visits,
sumIf(views, event_type = 1) as views,
lastAt as createdAt
from website_event_stats_hourly
where website_id = {websiteId:UUID}
${dateQuery}
${filterQuery}
group by session_id, website_id, hostname, browser, os, device, screen, language, country, subdivision1, city
order by lastAt desc
limit 1000)
select * from sessions
`,
params,
pageParams,
);
}

View file

@ -0,0 +1,71 @@
import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse';
import { runQuery, PRISMA, CLICKHOUSE } from 'lib/db';
import { QueryFilters } from 'lib/types';
export async function getWebsiteSessionsWeekly(
...args: [websiteId: string, filters?: QueryFilters]
) {
return runQuery({
[PRISMA]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args),
});
}
async function relationalQuery(websiteId: string, filters: QueryFilters) {
const { timezone = 'utc' } = filters;
const { rawQuery, getDateWeeklySQL, parseFilters } = prisma;
const { params } = await parseFilters(websiteId, filters);
return rawQuery(
`
select
${getDateWeeklySQL('created_at', timezone)} as time,
count(distinct session_id) as value
from website_event
where website_id = {{websiteId::uuid}}
and created_at between {{startDate}} and {{endDate}}
group by time
order by 2
`,
params,
).then(formatResults);
}
async function clickhouseQuery(websiteId: string, filters: QueryFilters) {
const { timezone = 'utc' } = filters;
const { rawQuery } = clickhouse;
const { startDate, endDate } = filters;
return rawQuery(
`
select
formatDateTime(toDateTime(created_at, '${timezone}'), '%w:%H') as time,
count(distinct session_id) as value
from website_event_stats_hourly
where website_id = {websiteId:UUID}
and created_at between {startDate:DateTime64} and {endDate:DateTime64}
group by time
order by time
`,
{ websiteId, startDate, endDate },
).then(formatResults);
}
function formatResults(data: any) {
const days = [];
for (let i = 0; i < 7; i++) {
days.push([]);
for (let j = 0; j < 24; j++) {
days[i].push(
Number(
data.find(({ time }) => time === `${i}:${j.toString().padStart(2, '0')}`)?.value || 0,
),
);
}
}
return days;
}

View file

@ -1,16 +1,16 @@
import clickhouse from 'lib/clickhouse';
import { DATA_TYPE } from 'lib/constants';
import { uuid } from 'lib/crypto';
import { flattenJSON, getStringValue } from 'lib/data';
import prisma from 'lib/prisma';
import { DynamicData } from 'lib/types';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import kafka from 'lib/kafka';
import prisma from 'lib/prisma';
import { DynamicData } from 'lib/types';
export async function saveSessionData(data: {
websiteId: string;
sessionId: string;
sessionData: DynamicData;
createdAt?: string;
}) {
return runQuery({
[PRISMA]: () => relationalQuery(data),
@ -77,11 +77,12 @@ async function clickhouseQuery(data: {
websiteId: string;
sessionId: string;
sessionData: DynamicData;
createdAt?: string;
}) {
const { websiteId, sessionId, sessionData, createdAt } = data;
const { websiteId, sessionId, sessionData } = data;
const { getDateFormat, sendMessages } = kafka;
const { insert, getUTCString } = clickhouse;
const { sendMessages } = kafka;
const createdAt = getUTCString();
const jsonKeys = flattenJSON(sessionData);
@ -93,12 +94,16 @@ async function clickhouseQuery(data: {
data_type: dataType,
string_value: getStringValue(value, dataType),
number_value: dataType === DATA_TYPE.number ? value : null,
date_value: dataType === DATA_TYPE.date ? getDateFormat(value) : null,
created_at: createdAt ?? getDateFormat(new Date()),
date_value: dataType === DATA_TYPE.date ? getUTCString(value) : null,
created_at: createdAt,
};
});
await sendMessages(messages, 'session_data');
if (kafka.enabled) {
await sendMessages('session_data', messages);
} else {
await insert('session_data', messages);
}
return data;
}