ambiser
/
umami
9
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: separate mongoQuery & add mongo filter

pull/2040/head
minkik 2023-05-19 14:10:53 +09:00 committed by Joseph Lee
parent 4c57ab1388
commit b5b689b156
15 changed files with 858 additions and 710 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
lib/db.js
View File

@ -26,9 +26,12 @@ export function getDatabaseType(url = process.env.DATABASE_URL) {
export async function runQuery(queries) { export async function runQuery(queries) {
const db = getDatabaseType(process.env.CLICKHOUSE_URL || process.env.DATABASE_URL); const db = getDatabaseType(process.env.CLICKHOUSE_URL || process.env.DATABASE_URL);
if (db === POSTGRESQL || db === MYSQL || db === MONGODB) { if (db === POSTGRESQL || db === MYSQL) {
return queries[PRISMA](); return queries[PRISMA]();
} }
if (db === MONGODB) {
return queries[MONGODB]();
}
if (db === CLICKHOUSE) { if (db === CLICKHOUSE) {
if (queries[KAFKA]) { if (queries[KAFKA]) {

20
lib/prisma.ts
View File

@ -142,6 +142,25 @@ function parseFilters(
}; };
} }
function parseMongoFilter(filters: { [key: string]: any } = {}) {
const query = {};
for (let k in filters) {
const v = filters[k];
if (v !== undefined) {
const tempK = FILTER_COLUMNS[k];
if (tempK !== undefined) {
k = tempK;
}
if (k === 'browser' || k === 'os' || k === 'device' || k === 'language') {
k = 'session.' + k;
}
query[k] = v;
}
}
return { $match: query };
}
async function rawQuery(query: string, params: never[] = []): Promise<any> { async function rawQuery(query: string, params: never[] = []): Promise<any> {
const db = getDatabaseType(process.env.DATABASE_URL); const db = getDatabaseType(process.env.DATABASE_URL);
@ -163,5 +182,6 @@ export default {
getEventDataFilterQuery, getEventDataFilterQuery,
toUuid, toUuid,
parseFilters, parseFilters,
parseMongoFilter,
rawQuery, rawQuery,
}; };

225
queries/analytics/event/getEventMetrics.ts
View File

@ -1,6 +1,6 @@
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse'; import clickhouse from 'lib/clickhouse';
import { runQuery, CLICKHOUSE, PRISMA } from 'lib/db'; import { runQuery, CLICKHOUSE, PRISMA, MONGODB } from 'lib/db';
import { WebsiteEventMetric } from 'lib/types'; import { WebsiteEventMetric } from 'lib/types';
import { EVENT_TYPE } from 'lib/constants'; import { EVENT_TYPE } from 'lib/constants';
import { loadWebsite } from 'lib/query'; import { loadWebsite } from 'lib/query';
@ -23,6 +23,7 @@ export async function getEventMetrics(
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(...args), [PRISMA]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args), [CLICKHOUSE]: () => clickhouseQuery(...args),
[MONGODB]: () => mongodbQuery(...args),
}); });
} }
@ -45,107 +46,13 @@ async function relationalQuery(
}; };
}, },
) { ) {
const { getDatabaseType, toUuid, rawQuery, getDateQuery, getFilterQuery, client } = prisma; const { toUuid, rawQuery, getDateQuery, getFilterQuery } = prisma;
const website = await loadWebsite(websiteId); const website = await loadWebsite(websiteId);
const resetDate = new Date(website?.resetAt || website?.createdAt); const resetDate = new Date(website?.resetAt || website?.createdAt);
const params: any = [websiteId, resetDate, startDate, endDate]; const params: any = [websiteId, resetDate, startDate, endDate];
const filterQuery = getFilterQuery(filters, params); const filterQuery = getFilterQuery(filters, params);
const db = getDatabaseType(); return rawQuery(
`select
if (db === 'mongodb') {
return await client.websiteEvent.aggregateRaw({
pipeline: [
{
$match: {
$expr: {
$and: [
{
$eq: ['$event_type', EVENT_TYPE.customEvent],
},
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: resetDate.toISOString(),
},
},
],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: startDate.toISOString(),
},
},
],
},
{
$lte: [
'$created_at',
{
$dateFromString: {
dateString: endDate.toISOString(),
},
},
],
},
],
},
},
},
{
$project: {
t: {
$dateTrunc: {
date: '$created_at',
unit: unit,
timezone: timezone,
},
},
event_name: 1,
},
},
{
$group: {
_id: {
t: '$t',
name: '$event_name',
},
y: {
$sum: 1,
},
},
},
{
$project: {
x: '$_id.name',
t: {
$dateToString: {
date: '$_id.t',
format: getDateQuery('', unit, timezone),
timezone: timezone,
},
},
y: 1,
_id: 0,
},
},
{
$sort: {
t: 1,
},
},
],
});
} else {
return rawQuery(
`select
event_name x, event_name x,
${getDateQuery('created_at', unit, timezone)} t, ${getDateQuery('created_at', unit, timezone)} t,
count(*) y count(*) y
@ -157,9 +64,8 @@ async function relationalQuery(
${filterQuery} ${filterQuery}
group by 1, 2 group by 1, 2
order by 2`, order by 2`,
params, params,
); );
}
} }
async function clickhouseQuery( async function clickhouseQuery(
@ -202,3 +108,120 @@ async function clickhouseQuery(
params, params,
); );
} }
async function mongodbQuery(
websiteId: string,
{
startDate,
endDate,
timezone = 'utc',
unit = 'day',
filters,
}: {
startDate: Date;
endDate: Date;
timezone: string;
unit: string;
filters: {
url: string;
eventName: string;
};
},
) {
const { getDateQuery, parseMongoFilter, client } = prisma;
const website = await loadWebsite(websiteId);
const resetDate = new Date(website?.resetAt || website?.createdAt);
const mongoFilter = parseMongoFilter(filters);
return await client.websiteEvent.aggregateRaw({
pipeline: [
mongoFilter,
{
$match: {
$expr: {
$and: [
{
$eq: ['$event_type', EVENT_TYPE.customEvent],
},
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: resetDate.toISOString(),
},
},
],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: startDate.toISOString(),
},
},
],
},
{
$lte: [
'$created_at',
{
$dateFromString: {
dateString: endDate.toISOString(),
},
},
],
},
],
},
},
},
{
$project: {
t: {
$dateTrunc: {
date: '$created_at',
unit: unit,
timezone: timezone,
},
},
event_name: 1,
},
},
{
$group: {
_id: {
t: '$t',
name: '$event_name',
},
y: {
$sum: 1,
},
},
},
{
$project: {
x: '$_id.name',
t: {
$dateToString: {
date: '$_id.t',
format: getDateQuery('', unit, timezone),
timezone: timezone,
},
},
y: 1,
_id: 0,
},
},
{
$sort: {
t: 1,
},
},
],
});
}

4
queries/analytics/event/getEvents.ts
View File

@ -1,11 +1,11 @@
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse'; import clickhouse from 'lib/clickhouse';
import { runQuery, CLICKHOUSE, PRISMA } from 'lib/db'; import { runQuery, CLICKHOUSE, PRISMA, MONGODB } from 'lib/db';
import { EVENT_TYPE } from 'lib/constants';
export function getEvents(...args: [websiteId: string, startAt: Date, eventType: number]) { export function getEvents(...args: [websiteId: string, startAt: Date, eventType: number]) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(...args), [PRISMA]: () => relationalQuery(...args),
[MONGODB]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args), [CLICKHOUSE]: () => clickhouseQuery(...args),
}); });
} }

3
queries/analytics/event/saveEvent.ts
View File

@ -1,5 +1,5 @@
import { EVENT_NAME_LENGTH, URL_LENGTH, EVENT_TYPE } from 'lib/constants'; import { EVENT_NAME_LENGTH, URL_LENGTH, EVENT_TYPE } from 'lib/constants';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db'; import { CLICKHOUSE, MONGODB, PRISMA, runQuery } from 'lib/db';
import kafka from 'lib/kafka'; import kafka from 'lib/kafka';
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
import { uuid } from 'lib/crypto'; import { uuid } from 'lib/crypto';
@ -29,6 +29,7 @@ export async function saveEvent(args: {
}) { }) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(args), [PRISMA]: () => relationalQuery(args),
[MONGODB]: () => relationalQuery(args),
[CLICKHOUSE]: () => clickhouseQuery(args), [CLICKHOUSE]: () => clickhouseQuery(args),
}); });
} }

258
queries/analytics/eventData/getEventData.ts
View File

@ -1,5 +1,5 @@
import clickhouse from 'lib/clickhouse'; import clickhouse from 'lib/clickhouse';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db'; import { CLICKHOUSE, MONGODB, PRISMA, runQuery } from 'lib/db';
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
import { WebsiteEventDataMetric } from 'lib/types'; import { WebsiteEventDataMetric } from 'lib/types';
import { loadWebsite } from 'lib/query'; import { loadWebsite } from 'lib/query';
@ -23,6 +23,7 @@ export async function getEventData(
): Promise<WebsiteEventDataMetric[]> { ): Promise<WebsiteEventDataMetric[]> {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(...args), [PRISMA]: () => relationalQuery(...args),
[MONGODB]: () => mongoQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args), [CLICKHOUSE]: () => clickhouseQuery(...args),
}); });
} }
@ -47,123 +48,13 @@ async function relationalQuery(
}, },
) { ) {
const { startDate, endDate, timeSeries, eventName, urlPath, filters } = data; const { startDate, endDate, timeSeries, eventName, urlPath, filters } = data;
const { getDatabaseType, toUuid, rawQuery, getEventDataFilterQuery, getDateQuery, client } = const { toUuid, rawQuery, getEventDataFilterQuery, getDateQuery } = prisma;
prisma;
const db = getDatabaseType();
const website = await loadWebsite(websiteId); const website = await loadWebsite(websiteId);
const resetDate = new Date(website?.resetAt || website?.createdAt); const resetDate = new Date(website?.resetAt || website?.createdAt);
const params: any = [websiteId, resetDate, startDate, endDate, eventName || '']; const params: any = [websiteId, resetDate, startDate, endDate, eventName || ''];
if (db === 'mongodb') { return rawQuery(
let joinAggregation: any = { match: {} }; `select
let matchAggregation: any = { match: {} };
let eventTypeProjectProperty = '';
let urlProjectProperty = '';
if (eventName || urlPath) {
joinAggregation = {
$lookup: {
from: 'website_event',
localField: 'website_event_id',
foreignField: '_id',
as: 'result',
},
};
eventTypeProjectProperty = 'event_name: {$arrayElemAt: ["$result.event_name", 0]}';
}
if (eventName) {
matchAggregation = {
$match: {
'result.event_name': eventName,
},
};
}
if (urlPath) {
urlProjectProperty = 'url_path: {$arrayElemAt: ["$result.url_path", 0],}';
}
let timeProjectProperty = '';
if (timeSeries) {
timeProjectProperty = `t: $dateTrunc: {date: "$created_at",unit: ${timeSeries.unit}, timezone : ${timeSeries.timezone}`;
}
return await client.websiteEvent.aggregateRaw({
pipeline: [
{
$match: {
$expr: {
$and: [
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: resetDate.toISOString(),
},
},
],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: startDate.toISOString(),
},
},
],
},
{
$lte: [
'$created_at',
{
$dateFromString: {
dateString: endDate.toISOString(),
},
},
],
},
],
},
},
},
joinAggregation,
matchAggregation,
{
$project: {
eventTypeProjectProperty,
timeProjectProperty,
urlProjectProperty,
},
},
{
$group: {
_id: {
url_path: '$url_path',
event_name: '$event_name',
t: '$t',
},
x: {
$sum: 1,
},
},
},
{
$project: {
url_path: '$_id.url_path',
urlPath: '$_id.url_path',
event_name: '$_id.event_name',
eventName: '$_id.event_name',
x: 1,
t: '$_id.t',
_id: 0,
},
},
],
});
} else {
return rawQuery(
`select
count(*) x count(*) x
${eventName ? `,event_name eventName` : ''} ${eventName ? `,event_name eventName` : ''}
${urlPath ? `,url_path urlPath` : ''} ${urlPath ? `,url_path urlPath` : ''}
@ -182,9 +73,8 @@ async function relationalQuery(
${eventName ? `and eventName = $5` : ''} ${eventName ? `and eventName = $5` : ''}
${getEventDataFilterQuery(filters, params)} ${getEventDataFilterQuery(filters, params)}
${timeSeries ? 'group by t' : ''}`, ${timeSeries ? 'group by t' : ''}`,
params, params,
); );
}
} }
async function clickhouseQuery( async function clickhouseQuery(
@ -231,3 +121,137 @@ async function clickhouseQuery(
params, params,
); );
} }
async function mongoQuery(
websiteId: string,
data: {
startDate: Date;
endDate: Date;
timeSeries?: {
unit: string;
timezone: string;
};
eventName: string;
urlPath?: string;
filters: [
{
eventKey?: string;
eventValue?: string | number | boolean | Date;
},
];
},
) {
const { startDate, endDate, timeSeries, eventName, urlPath, filters } = data;
const { client, parseMongoFilter } = prisma;
const website = await loadWebsite(websiteId);
const resetDate = new Date(website?.resetAt || website?.createdAt);
const mongoFilter = parseMongoFilter(filters);
let joinAggregation: any = { match: {} };
let matchAggregation: any = { match: {} };
let eventTypeProjectProperty = '';
let urlProjectProperty = '';
if (eventName || urlPath) {
joinAggregation = {
$lookup: {
from: 'website_event',
localField: 'website_event_id',
foreignField: '_id',
as: 'result',
},
};
eventTypeProjectProperty = 'event_name: {$arrayElemAt: ["$result.event_name", 0]}';
}
if (eventName) {
matchAggregation = {
$match: {
'result.event_name': eventName,
},
};
}
if (urlPath) {
urlProjectProperty = 'url_path: {$arrayElemAt: ["$result.url_path", 0],}';
}
let timeProjectProperty = '';
if (timeSeries) {
timeProjectProperty = `t: $dateTrunc: {date: "$created_at",unit: ${timeSeries.unit}, timezone : ${timeSeries.timezone}`;
}
return await client.websiteEvent.aggregateRaw({
pipeline: [
mongoFilter,
{
$match: {
$expr: {
$and: [
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: resetDate.toISOString(),
},
},
],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: startDate.toISOString(),
},
},
],
},
{
$lte: [
'$created_at',
{
$dateFromString: {
dateString: endDate.toISOString(),
},
},
],
},
],
},
},
},
joinAggregation,
matchAggregation,
{
$project: {
eventTypeProjectProperty,
timeProjectProperty,
urlProjectProperty,
},
},
{
$group: {
_id: {
url_path: '$url_path',
event_name: '$event_name',
t: '$t',
},
x: {
$sum: 1,
},
},
},
{
$project: {
url_path: '$_id.url_path',
urlPath: '$_id.url_path',
event_name: '$_id.event_name',
eventName: '$_id.event_name',
x: 1,
t: '$_id.t',
_id: 0,
},
},
],
});
}

3
queries/analytics/eventData/saveEventData.ts
View File

@ -1,7 +1,7 @@
import { Prisma } from '@prisma/client'; import { Prisma } from '@prisma/client';
import { EVENT_DATA_TYPE } from 'lib/constants'; import { EVENT_DATA_TYPE } from 'lib/constants';
import { uuid } from 'lib/crypto'; import { uuid } from 'lib/crypto';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db'; import { CLICKHOUSE, MONGODB, PRISMA, runQuery } from 'lib/db';
import { flattenJSON } from 'lib/eventData'; import { flattenJSON } from 'lib/eventData';
import kafka from 'lib/kafka'; import kafka from 'lib/kafka';
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
@ -18,6 +18,7 @@ export async function saveEventData(args: {
}) { }) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(args), [PRISMA]: () => relationalQuery(args),
[MONGODB]: () => relationalQuery(args),
[CLICKHOUSE]: () => clickhouseQuery(args), [CLICKHOUSE]: () => clickhouseQuery(args),
}); });
} }

209
queries/analytics/pageview/getPageviewMetrics.ts
View File

@ -1,6 +1,6 @@
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse'; import clickhouse from 'lib/clickhouse';
import { runQuery, CLICKHOUSE, PRISMA } from 'lib/db'; import { runQuery, CLICKHOUSE, PRISMA, MONGODB } from 'lib/db';
import { EVENT_TYPE } from 'lib/constants'; import { EVENT_TYPE } from 'lib/constants';
import { loadWebsite } from 'lib/query'; import { loadWebsite } from 'lib/query';
@ -17,6 +17,7 @@ export async function getPageviewMetrics(
) { ) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(...args), [PRISMA]: () => relationalQuery(...args),
[MONGODB]: () => mongodbQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args), [CLICKHOUSE]: () => clickhouseQuery(...args),
}); });
} }
@ -31,8 +32,7 @@ async function relationalQuery(
}, },
) { ) {
const { startDate, endDate, filters = {}, column } = criteria; const { startDate, endDate, filters = {}, column } = criteria;
const { getDatabaseType, rawQuery, parseFilters, toUuid, client } = prisma; const { rawQuery, parseFilters, toUuid } = prisma;
const db = getDatabaseType();
const website = await loadWebsite(websiteId); const website = await loadWebsite(websiteId);
const resetDate = new Date(website?.resetAt || website?.createdAt); const resetDate = new Date(website?.resetAt || website?.createdAt);
const params: any = [ const params: any = [
@ -44,99 +44,15 @@ async function relationalQuery(
]; ];
let excludeDomain = ''; let excludeDomain = '';
let excludeDomainMongo = {};
if (column === 'referrer_domain') { if (column === 'referrer_domain') {
excludeDomain = 'and website_event.referrer_domain != $6'; excludeDomain = 'and website_event.referrer_domain != $6';
excludeDomainMongo = {
$ne: ['$referrer_domain', website.domain],
};
params.push(website.domain); params.push(website.domain);
} }
const { filterQuery, joinSession } = parseFilters(filters, params); const { filterQuery, joinSession } = parseFilters(filters, params);
return rawQuery(
if (db === 'mongodb') { `select ${column} x, count(*) y
return await client.websiteEvent.aggregateRaw({
pipeline: [
{
$match: {
$expr: {
$and: [
{
$eq: ['$event_type', params[4]],
},
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: resetDate.toISOString(),
},
},
],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: startDate.toISOString(),
},
},
],
},
{
$lte: [
'$created_at',
{
$dateFromString: {
dateString: endDate.toISOString(),
},
},
],
},
excludeDomainMongo,
],
},
},
},
{
$group: {
_id: '$' + column,
y: {
$sum: 1,
},
},
},
{
$project: {
x: '$_id',
y: 1,
_id: 0,
},
},
{
$sort: {
x: 1,
},
},
{
$sort: {
y: -1,
},
},
{
$limit: 100,
},
],
});
} else {
return rawQuery(
`select ${column} x, count(*) y
from website_event from website_event
${joinSession} ${joinSession}
where website_event.website_id = $1${toUuid()} where website_event.website_id = $1${toUuid()}
@ -148,9 +64,8 @@ async function relationalQuery(
group by 1 group by 1
order by 2 desc order by 2 desc
limit 100`, limit 100`,
params, params,
); );
}
} }
async function clickhouseQuery( async function clickhouseQuery(
@ -196,3 +111,113 @@ async function clickhouseQuery(
params, params,
); );
} }
async function mongodbQuery(
websiteId: string,
criteria: {
startDate: Date;
endDate: Date;
column: string;
filters: object;
},
) {
const { startDate, endDate, filters = {}, column } = criteria;
const { parseMongoFilter, client } = prisma;
const website = await loadWebsite(websiteId);
const resetDate = new Date(website?.resetAt || website?.createdAt);
const params: any = [
websiteId,
resetDate,
startDate,
endDate,
column === 'event_name' ? EVENT_TYPE.customEvent : EVENT_TYPE.pageView,
];
let excludeDomainMongo: any = '';
if (column === 'referrer_domain') {
excludeDomainMongo = {
$ne: ['$referrer_domain', website.domain],
};
params.push(website.domain);
}
const mongoFilter = parseMongoFilter(filters);
return await client.websiteEvent.aggregateRaw({
pipeline: [
mongoFilter,
{
$match: {
$expr: {
$and: [
{
$eq: ['$event_type', params[4]],
},
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: resetDate.toISOString(),
},
},
],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: startDate.toISOString(),
},
},
],
},
{
$lte: [
'$created_at',
{
$dateFromString: {
dateString: endDate.toISOString(),
},
},
],
},
excludeDomainMongo,
],
},
},
},
{
$group: {
_id: '$' + column,
y: {
$sum: 1,
},
},
},
{
$project: {
x: '$_id',
y: 1,
_id: 0,
},
},
{
$sort: {
x: 1,
},
},
{
$sort: {
y: -1,
},
},
{
$limit: 100,
},
],
});
}

272
queries/analytics/pageview/getPageviewStats.ts
View File

@ -1,5 +1,5 @@
import clickhouse from 'lib/clickhouse'; import clickhouse from 'lib/clickhouse';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db'; import { CLICKHOUSE, MONGODB, PRISMA, runQuery } from 'lib/db';
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
import { EVENT_TYPE } from 'lib/constants'; import { EVENT_TYPE } from 'lib/constants';
import { loadWebsite } from 'lib/query'; import { loadWebsite } from 'lib/query';
@ -20,6 +20,7 @@ export async function getPageviewStats(
) { ) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(...args), [PRISMA]: () => relationalQuery(...args),
[MONGODB]: () => mongodbQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args), [CLICKHOUSE]: () => clickhouseQuery(...args),
}); });
} }
@ -45,126 +46,14 @@ async function relationalQuery(
filters = {}, filters = {},
sessionKey = 'session_id', sessionKey = 'session_id',
} = criteria; } = criteria;
const { getDatabaseType, toUuid, getDateQuery, parseFilters, rawQuery, client } = prisma; const { toUuid, getDateQuery, parseFilters, rawQuery } = prisma;
const db = getDatabaseType();
const website = await loadWebsite(websiteId); const website = await loadWebsite(websiteId);
const resetDate = new Date(website?.resetAt || website?.createdAt); const resetDate = new Date(website?.resetAt || website?.createdAt);
const params: any = [websiteId, resetDate, startDate, endDate]; const params: any = [websiteId, resetDate, startDate, endDate];
const { filterQuery, joinSession } = parseFilters(filters, params); const { filterQuery, joinSession } = parseFilters(filters, params);
let sessionInclude = ''; return rawQuery(
let sessionGroupAggregation: any = { $match: {} }; `select ${getDateQuery('website_event.created_at', unit, timezone)} x,
let sessionProjectAggregation: any = { $match: {} };
if (count !== '*') {
sessionInclude = 'session_id : 1';
sessionGroupAggregation = {
$group: {
_id: {
t: '$t',
session_id: '$session_id',
},
},
};
sessionProjectAggregation = {
$project: {
t: '$_id.t',
},
};
}
if (db === 'mongodb') {
return await client.websiteEvent.aggregateRaw({
pipeline: [
{
$match: {
$expr: {
$and: [
{
$eq: ['$event_type', EVENT_TYPE.pageView],
},
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: resetDate.toISOString(),
},
},
],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: startDate.toISOString(),
},
},
],
},
{
$lte: [
'$created_at',
{
$dateFromString: {
dateString: endDate.toISOString(),
},
},
],
},
],
},
},
},
{
$project: {
t: {
$dateTrunc: {
date: '$created_at',
unit: unit,
},
},
sessionInclude,
},
},
sessionGroupAggregation,
sessionProjectAggregation,
{
$group: {
_id: {
t: '$t',
session_id: '$session_id',
},
y: {
$sum: 1,
},
},
},
{
$project: {
x: {
$dateToString: {
date: '$_id.t',
format: getDateQuery('', unit, timezone),
timezone: timezone,
},
},
y: 1,
_id: 0,
},
},
{
$sort: {
x: 1,
},
},
],
});
} else {
return rawQuery(
`select ${getDateQuery('website_event.created_at', unit, timezone)} x,
count(${count !== '*' ? `${count}${sessionKey}` : count}) y count(${count !== '*' ? `${count}${sessionKey}` : count}) y
from website_event from website_event
${joinSession} ${joinSession}
@ -174,9 +63,8 @@ async function relationalQuery(
and event_type = ${EVENT_TYPE.pageView} and event_type = ${EVENT_TYPE.pageView}
${filterQuery} ${filterQuery}
group by 1`, group by 1`,
params, params,
); );
}
} }
async function clickhouseQuery( async function clickhouseQuery(
@ -231,3 +119,149 @@ async function clickhouseQuery(
params, params,
); );
} }
async function mongodbQuery(
websiteId: string,
criteria: {
startDate: Date;
endDate: Date;
timezone?: string;
unit?: string;
count?: string;
filters: object;
sessionKey?: string;
},
) {
const {
startDate,
endDate,
timezone = 'utc',
unit = 'day',
count = '*',
filters = {},
} = criteria;
const { getDateQuery, client, parseMongoFilter } = prisma;
const website = await loadWebsite(websiteId);
const resetDate = new Date(website?.resetAt || website?.createdAt);
const mongoFilter = parseMongoFilter(filters);
let sessionInclude = '';
let sessionGroupAggregation: any = { $match: {} };
const sessionLookUpAggregation: any = {
$lookup: {
from: 'session',
foreignField: '_id',
localField: 'session_id',
as: 'session',
},
};
let sessionProjectAggregation: any = { $match: {} };
if (count !== '*') {
sessionInclude = 'session_id : 1';
sessionGroupAggregation = {
$group: {
_id: {
t: '$t',
session_id: '$session_id',
},
},
};
sessionProjectAggregation = {
$project: {
t: '$_id.t',
},
};
}
return await client.websiteEvent.aggregateRaw({
pipeline: [
sessionLookUpAggregation,
mongoFilter,
{
$match: {
$expr: {
$and: [
{
$eq: ['$event_type', EVENT_TYPE.pageView],
},
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: resetDate.toISOString(),
},
},
],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: startDate.toISOString(),
},
},
],
},
{
$lte: [
'$created_at',
{
$dateFromString: {
dateString: endDate.toISOString(),
},
},
],
},
],
},
},
},
{
$project: {
t: {
$dateTrunc: {
date: '$created_at',
unit: unit,
},
},
sessionInclude,
},
},
sessionGroupAggregation,
sessionProjectAggregation,
{
$group: {
_id: {
t: '$t',
session_id: '$session_id',
},
y: {
$sum: 1,
},
},
},
{
$project: {
x: {
$dateToString: {
date: '$_id.t',
format: getDateQuery('', unit, timezone),
timezone: timezone,
},
},
y: 1,
_id: 0,
},
},
{
$sort: {
x: 1,
},
},
],
});
}

3
queries/analytics/session/createSession.ts
View File

@ -1,4 +1,4 @@
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db'; import { CLICKHOUSE, MONGODB, PRISMA, runQuery } from 'lib/db';
import kafka from 'lib/kafka'; import kafka from 'lib/kafka';
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
import cache from 'lib/cache'; import cache from 'lib/cache';
@ -7,6 +7,7 @@ import { Prisma } from '@prisma/client';
export async function createSession(args: Prisma.SessionCreateInput) { export async function createSession(args: Prisma.SessionCreateInput) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(args), [PRISMA]: () => relationalQuery(args),
[MONGODB]: () => relationalQuery(args),
[CLICKHOUSE]: () => clickhouseQuery(args), [CLICKHOUSE]: () => clickhouseQuery(args),
}).then(async data => { }).then(async data => {
if (cache.enabled) { if (cache.enabled) {

3
queries/analytics/session/getSession.ts
View File

@ -1,11 +1,12 @@
import clickhouse from 'lib/clickhouse'; import clickhouse from 'lib/clickhouse';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db'; import { CLICKHOUSE, MONGODB, PRISMA, runQuery } from 'lib/db';
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
import { Prisma } from '@prisma/client'; import { Prisma } from '@prisma/client';
export async function getSession(args: { id: string }) { export async function getSession(args: { id: string }) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(args), [PRISMA]: () => relationalQuery(args),
[MONGODB]: () => relationalQuery(args),
[CLICKHOUSE]: () => clickhouseQuery(args), [CLICKHOUSE]: () => clickhouseQuery(args),
}); });
} }

205
queries/analytics/session/getSessionMetrics.ts
View File

@ -1,6 +1,6 @@
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse'; import clickhouse from 'lib/clickhouse';
import { runQuery, CLICKHOUSE, PRISMA } from 'lib/db'; import { runQuery, CLICKHOUSE, PRISMA, MONGODB } from 'lib/db';
import { EVENT_TYPE } from 'lib/constants'; import { EVENT_TYPE } from 'lib/constants';
import { loadWebsite } from 'lib/query'; import { loadWebsite } from 'lib/query';
@ -12,6 +12,7 @@ export async function getSessionMetrics(
) { ) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(...args), [PRISMA]: () => relationalQuery(...args),
[MONGODB]: () => mongodbQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args), [CLICKHOUSE]: () => clickhouseQuery(...args),
}); });
} }
@ -23,103 +24,12 @@ async function relationalQuery(
const website = await loadWebsite(websiteId); const website = await loadWebsite(websiteId);
const resetDate = new Date(website?.resetAt || website?.createdAt); const resetDate = new Date(website?.resetAt || website?.createdAt);
const { startDate, endDate, column, filters = {} } = criteria; const { startDate, endDate, column, filters = {} } = criteria;
const { getDatabaseType, toUuid, parseFilters, rawQuery, client } = prisma; const { toUuid, parseFilters, rawQuery } = prisma;
const db = getDatabaseType();
const params: any = [websiteId, resetDate, startDate, endDate]; const params: any = [websiteId, resetDate, startDate, endDate];
const { filterQuery, joinSession } = parseFilters(filters, params); const { filterQuery, joinSession } = parseFilters(filters, params);
if (db === 'mongodb') { return rawQuery(
return await client.websiteEvent.aggregateRaw({ `select ${column} x, count(*) y
pipeline: [
{
$match: {
$expr: {
$and: [
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: resetDate.toISOString(),
},
},
],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: startDate.toISOString(),
},
},
],
},
{
$lte: [
'$created_at',
{
$dateFromString: {
dateString: endDate.toISOString(),
},
},
],
},
],
},
},
},
{
$group: {
_id: '$session_id',
},
},
{
$lookup: {
from: 'session',
localField: '_id',
foreignField: '_id',
as: 'session',
},
},
{
$project: {
session: {
$arrayElemAt: ['$session', 0],
},
},
},
{
$group: {
_id: '$session.' + column,
sum: {
$sum: 1,
},
},
},
{
$project: {
x: '$_id',
y: '$sum',
_id: 0,
},
},
{
$sort: {
sum: -1,
},
},
{
$limit: 100,
},
],
});
} else {
return rawQuery(
`select ${column} x, count(*) y
from session as x from session as x
where x.session_id in ( where x.session_id in (
select website_event.session_id select website_event.session_id
@ -135,9 +45,8 @@ async function relationalQuery(
group by 1 group by 1
order by 2 desc order by 2 desc
limit 100`, limit 100`,
params, params,
); );
}
} }
async function clickhouseQuery( async function clickhouseQuery(
websiteId: string, websiteId: string,
@ -164,3 +73,103 @@ async function clickhouseQuery(
params, params,
); );
} }
async function mongodbQuery(
websiteId: string,
criteria: { startDate: Date; endDate: Date; column: string; filters: object },
) {
const website = await loadWebsite(websiteId);
const resetDate = new Date(website?.resetAt || website?.createdAt);
const { startDate, endDate, column, filters = {} } = criteria;
const { parseMongoFilter, client } = prisma;
const mongoFilter = parseMongoFilter(filters);
return await client.websiteEvent.aggregateRaw({
pipeline: [
mongoFilter,
{
$match: {
$expr: {
$and: [
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: resetDate.toISOString(),
},
},
],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: startDate.toISOString(),
},
},
],
},
{
$lte: [
'$created_at',
{
$dateFromString: {
dateString: endDate.toISOString(),
},
},
],
},
],
},
},
},
{
$group: {
_id: '$session_id',
},
},
{
$lookup: {
from: 'session',
localField: '_id',
foreignField: '_id',
as: 'session',
},
},
{
$project: {
session: {
$arrayElemAt: ['$session', 0],
},
},
},
{
$group: {
_id: '$session.' + column,
sum: {
$sum: 1,
},
},
},
{
$project: {
x: '$_id',
y: '$sum',
_id: 0,
},
},
{
$sort: {
sum: -1,
},
},
{
$limit: 100,
},
],
});
}

3
queries/analytics/session/getSessions.ts
View File

@ -1,10 +1,11 @@
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse'; import clickhouse from 'lib/clickhouse';
import { runQuery, PRISMA, CLICKHOUSE } from 'lib/db'; import { runQuery, PRISMA, CLICKHOUSE, MONGODB } from 'lib/db';
export async function getSessions(...args: [websiteId: string, startAt: Date]) { export async function getSessions(...args: [websiteId: string, startAt: Date]) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(...args), [PRISMA]: () => relationalQuery(...args),
[MONGODB]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args), [CLICKHOUSE]: () => clickhouseQuery(...args),
}); });
} }

100
queries/analytics/stats/getActiveVisitors.ts
View File

@ -1,72 +1,31 @@
import { subMinutes } from 'date-fns'; import { subMinutes } from 'date-fns';
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse'; import clickhouse from 'lib/clickhouse';
import { runQuery, CLICKHOUSE, PRISMA } from 'lib/db'; import { runQuery, CLICKHOUSE, PRISMA, MONGODB } from 'lib/db';
export async function getActiveVisitors(...args: [websiteId: string]) { export async function getActiveVisitors(...args: [websiteId: string]) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(...args), [PRISMA]: () => relationalQuery(...args),
[MONGODB]: () => mongodbQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args), [CLICKHOUSE]: () => clickhouseQuery(...args),
}); });
} }
async function relationalQuery(websiteId: string) { async function relationalQuery(websiteId: string) {
const { getDatabaseType, toUuid, rawQuery, client } = prisma; const { toUuid, rawQuery } = prisma;
const db = getDatabaseType();
const date = subMinutes(new Date(), 5); const date = subMinutes(new Date(), 5);
const params: any = [websiteId, date]; const params: any = [websiteId, date];
if (db === 'mongodb') { return rawQuery(
const result: any = await client.websiteEvent.aggregateRaw({ `select count(distinct session_id) x
pipeline: [
{
$match: {
$expr: {
$and: [
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: date.toISOString(),
},
},
],
},
],
},
},
},
{
$group: {
_id: '$session_id',
},
},
{
$count: 'x',
},
],
});
if (result.length > 0) {
return { x: result[0].x };
} else {
return { x: 0 };
}
} else {
return rawQuery(
`select count(distinct session_id) x
from website_event from website_event
join website join website
on website_event.website_id = website.website_id on website_event.website_id = website.website_id
where website.website_id = $1${toUuid()} where website.website_id = $1${toUuid()}
and website_event.created_at >= $2`, and website_event.created_at >= $2`,
params, params,
); );
}
} }
async function clickhouseQuery(websiteId: string) { async function clickhouseQuery(websiteId: string) {
@ -81,3 +40,48 @@ async function clickhouseQuery(websiteId: string) {
params, params,
); );
} }
async function mongodbQuery(websiteId: string) {
const { client } = prisma;
const date = subMinutes(new Date(), 5);
const result: any = await client.websiteEvent.aggregateRaw({
pipeline: [
{
$match: {
$expr: {
$and: [
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: date.toISOString(),
},
},
],
},
],
},
},
},
{
$group: {
_id: '$session_id',
},
},
{
$count: 'x',
},
],
});
if (result.length > 0) {
return { x: result[0].x };
} else {
return { x: 0 };
}
}

255
queries/analytics/stats/getWebsiteStats.ts
View File

@ -1,6 +1,6 @@
import prisma from 'lib/prisma'; import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse'; import clickhouse from 'lib/clickhouse';
import { runQuery, CLICKHOUSE, PRISMA } from 'lib/db'; import { runQuery, CLICKHOUSE, PRISMA, MONGODB } from 'lib/db';
import { EVENT_TYPE } from 'lib/constants'; import { EVENT_TYPE } from 'lib/constants';
import { loadWebsite } from 'lib/query'; import { loadWebsite } from 'lib/query';
@ -12,6 +12,7 @@ export async function getWebsiteStats(
) { ) {
return runQuery({ return runQuery({
[PRISMA]: () => relationalQuery(...args), [PRISMA]: () => relationalQuery(...args),
[MONGODB]: () => mongodbQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args), [CLICKHOUSE]: () => clickhouseQuery(...args),
}); });
} }
@ -21,133 +22,13 @@ async function relationalQuery(
criteria: { startDate: Date; endDate: Date; filters: object }, criteria: { startDate: Date; endDate: Date; filters: object },
) { ) {
const { startDate, endDate, filters = {} } = criteria; const { startDate, endDate, filters = {} } = criteria;
const { const { toUuid, getDateQuery, getTimestampInterval, parseFilters, rawQuery } = prisma;
getDatabaseType,
toUuid,
getDateQuery,
getTimestampInterval,
parseFilters,
rawQuery,
client,
} = prisma;
const db = getDatabaseType();
const website = await loadWebsite(websiteId); const website = await loadWebsite(websiteId);
const resetDate = new Date(website?.resetAt || website?.createdAt); const resetDate = new Date(website?.resetAt || website?.createdAt);
const params: any = [websiteId, resetDate, startDate, endDate]; const params: any = [websiteId, resetDate, startDate, endDate];
const { filterQuery, joinSession } = parseFilters(filters, params); const { filterQuery, joinSession } = parseFilters(filters, params);
return rawQuery(
if (db === 'mongodb') { `select sum(t.c) as "pageviews",
return await client.websiteEvent.aggregateRaw({
pipeline: [
{
$match: {
$expr: {
$and: [
{
$eq: ['$event_type', EVENT_TYPE.pageView],
},
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: resetDate.toISOString(),
},
},
],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: startDate.toISOString(),
},
},
],
},
{
$lte: [
'$created_at',
{
$dateFromString: {
dateString: endDate.toISOString(),
},
},
],
},
],
},
},
},
{
$project: {
session_id: '$session_id',
hour: {
$toString: { $hour: '$created_at' },
},
created_at: '$created_at',
},
},
{
$group: {
_id: {
$concat: ['$session_id', ':', '$hour'],
},
session_id: { $first: '$session_id' },
hour: { $first: '$hour' },
count: { $sum: 1 },
timeMax: { $max: '$created_at' },
timeMin: { $min: '$created_at' },
},
},
{
$project: {
_id: '$_id',
session_id: '$session_id',
hour: '$hour',
count: '$count',
time: {
$dateDiff: {
endDate: '$timeMax',
startDate: '$timeMin',
unit: 'second',
},
},
bounce: {
$cond: {
if: { $eq: ['$count', 1] },
then: 1,
else: 0,
},
},
},
},
{
$group: {
_id: '$session_id',
pageviews: { $sum: '$count' },
bounces: { $sum: '$bounce' },
totaltime: { $sum: '$time' },
},
},
{
$group: {
_id: '',
pageviews: { $sum: '$pageviews' },
uniques: { $sum: 1 },
bounces: { $sum: '$bounces' },
totaltime: { $sum: '$totaltime' },
},
},
],
});
} else {
return rawQuery(
`select sum(t.c) as "pageviews",
count(distinct t.session_id) as "uniques", count(distinct t.session_id) as "uniques",
sum(case when t.c = 1 then 1 else 0 end) as "bounces", sum(case when t.c = 1 then 1 else 0 end) as "bounces",
sum(t.time) as "totaltime" sum(t.time) as "totaltime"
@ -167,9 +48,8 @@ async function relationalQuery(
${filterQuery} ${filterQuery}
group by 1, 2 group by 1, 2
) t`, ) t`,
params, params,
); );
}
} }
async function clickhouseQuery( async function clickhouseQuery(
@ -206,3 +86,124 @@ async function clickhouseQuery(
params, params,
); );
} }
async function mongodbQuery(
websiteId: string,
criteria: { startDate: Date; endDate: Date; filters: object },
) {
const { startDate, endDate, filters = {} } = criteria;
const { parseMongoFilter, client } = prisma;
const website = await loadWebsite(websiteId);
const resetDate = new Date(website?.resetAt || website?.createdAt);
const mongoFilter = parseMongoFilter(filters);
return await client.websiteEvent.aggregateRaw({
pipeline: [
mongoFilter,
{
$match: {
$expr: {
$and: [
{
$eq: ['$event_type', EVENT_TYPE.pageView],
},
{
$eq: ['$website_id', websiteId],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: resetDate.toISOString(),
},
},
],
},
{
$gte: [
'$created_at',
{
$dateFromString: {
dateString: startDate.toISOString(),
},
},
],
},
{
$lte: [
'$created_at',
{
$dateFromString: {
dateString: endDate.toISOString(),
},
},
],
},
],
},
},
},
{
$project: {
session_id: '$session_id',
hour: {
$toString: { $hour: '$created_at' },
},
created_at: '$created_at',
},
},
{
$group: {
_id: {
$concat: ['$session_id', ':', '$hour'],
},
session_id: { $first: '$session_id' },
hour: { $first: '$hour' },
count: { $sum: 1 },
timeMax: { $max: '$created_at' },
timeMin: { $min: '$created_at' },
},
},
{
$project: {
_id: '$_id',
session_id: '$session_id',
hour: '$hour',
count: '$count',
time: {
$dateDiff: {
endDate: '$timeMax',
startDate: '$timeMin',
unit: 'second',
},
},
bounce: {
$cond: {
if: { $eq: ['$count', 1] },
then: 1,
else: 0,
},
},
},
},
{
$group: {
_id: '$session_id',
pageviews: { $sum: '$count' },
bounces: { $sum: '$bounce' },
totaltime: { $sum: '$time' },
},
},
{
$group: {
_id: '',
pageviews: { $sum: '$pageviews' },
uniques: { $sum: 1 },
bounces: { $sum: '$bounces' },
totaltime: { $sum: '$totaltime' },
},
},
],
});
}