From 46b4b98d407cd3958d0eeadb94b59613ff59931b Mon Sep 17 00:00:00 2001 From: Brian Cao Date: Thu, 25 Aug 2022 11:07:47 -0700 Subject: [PATCH 1/8] update kafka messaging --- lib/db.js | 2 +- lib/{ => db}/kafka.js | 27 +++++++++++++++------- pages/api/heartbeat.js | 5 ++++ queries/analytics/event/saveEvent.js | 4 ++-- queries/analytics/pageview/savePageView.js | 4 ++-- queries/analytics/session/createSession.js | 4 ++-- 6 files changed, 31 insertions(+), 15 deletions(-) rename lib/{ => db}/kafka.js (69%) create mode 100644 pages/api/heartbeat.js diff --git a/lib/db.js b/lib/db.js index ff62da03..30fa79f6 100644 --- a/lib/db.js +++ b/lib/db.js @@ -2,7 +2,7 @@ import { PrismaClient } from '@prisma/client'; import { ClickHouse } from 'clickhouse'; import dateFormat from 'dateformat'; import chalk from 'chalk'; -import { getKafkaService } from './kafka'; +import { getKafkaService } from './db/kafka'; import { MYSQL, MYSQL_DATE_FORMATS, diff --git a/lib/kafka.js b/lib/db/kafka.js similarity index 69% rename from lib/kafka.js rename to lib/db/kafka.js index 988062b9..e2256d27 100644 --- a/lib/kafka.js +++ b/lib/db/kafka.js @@ -1,4 +1,4 @@ -import { Kafka } from 'kafkajs'; +import { Kafka, logLevel } from 'kafkajs'; import dateFormat from 'dateformat'; export function getKafkaClient() { @@ -14,6 +14,7 @@ export function getKafkaClient() { clientId: 'umami', brokers: brokers, connectionTimeout: 3000, + logLevel: logLevel.ERROR, }); } else { return new Kafka({ @@ -29,20 +30,29 @@ export function getKafkaClient() { }); } } - const kafka = global.kafka || getKafkaClient(); +let kafkaProducer = null; -if (process.env.NODE_ENV !== 'production') { - global.kafka = kafka; -} +(async () => { + kafkaProducer = global.kakfaProducer || (await getKafkaProducer()); -export { kafka }; + if (process.env.NODE_ENV !== 'production') { + global.kafka = kafka; + global.kakfaProducer = kafkaProducer; + } +})(); -export async function kafkaProducer(params, topic) { +export { kafka, kafkaProducer }; + +export async function getKafkaProducer() { const producer = kafka.producer(); await producer.connect(); - await producer.send({ + return producer; +} + +export async function sendKafkaMessage(params, topic) { + await kafkaProducer.send({ topic, messages: [ { @@ -50,6 +60,7 @@ export async function kafkaProducer(params, topic) { value: JSON.stringify(params), }, ], + acks: 0, }); } diff --git a/pages/api/heartbeat.js b/pages/api/heartbeat.js new file mode 100644 index 00000000..a53213ad --- /dev/null +++ b/pages/api/heartbeat.js @@ -0,0 +1,5 @@ +import { ok } from 'lib/response'; + +export default async (req, res) => { + return ok(res, 'nice'); +}; diff --git a/queries/analytics/event/saveEvent.js b/queries/analytics/event/saveEvent.js index 0786042c..ff4d5256 100644 --- a/queries/analytics/event/saveEvent.js +++ b/queries/analytics/event/saveEvent.js @@ -6,7 +6,7 @@ import { runAnalyticsQuery, runQuery, } from 'lib/db'; -import { kafkaProducer, getDateFormatKafka } from 'lib/kafka'; +import { sendKafkaMessage, getDateFormatKafka } from 'lib/db/kafka'; export async function saveEvent(...args) { return runAnalyticsQuery({ @@ -66,5 +66,5 @@ async function kafkaQuery(website_id, { event_uuid, session_uuid, url, event_nam event_name: event_name?.substr(0, 50), }; - await kafkaProducer(params, 'event'); + await sendKafkaMessage(params, 'event'); } diff --git a/queries/analytics/pageview/savePageView.js b/queries/analytics/pageview/savePageView.js index 0dd93b44..af0d88f6 100644 --- a/queries/analytics/pageview/savePageView.js +++ b/queries/analytics/pageview/savePageView.js @@ -6,7 +6,7 @@ import { runAnalyticsQuery, runQuery, } from 'lib/db'; -import { kafkaProducer, getDateFormatKafka } from 'lib/kafka'; +import { sendKafkaMessage, getDateFormatKafka } from 'lib/db/kafka'; export async function savePageView(...args) { return runAnalyticsQuery({ @@ -54,5 +54,5 @@ async function kafkaQuery(website_id, { session_uuid, url, referrer }) { referrer: referrer?.substr(0, URL_LENGTH), }; - await kafkaProducer(params, 'pageview'); + await sendKafkaMessage(params, 'pageview'); } diff --git a/queries/analytics/session/createSession.js b/queries/analytics/session/createSession.js index 5c4b982c..565d45a2 100644 --- a/queries/analytics/session/createSession.js +++ b/queries/analytics/session/createSession.js @@ -6,7 +6,7 @@ import { runAnalyticsQuery, runQuery, } from 'lib/db'; -import { kafkaProducer, getDateFormatKafka } from 'lib/kafka'; +import { sendKafkaMessage, getDateFormatKafka } from 'lib/db/kafka'; import { getSessionByUuid } from 'queries'; export async function createSession(...args) { @@ -73,7 +73,7 @@ async function kafkaQuery( country: country ? country : null, }; - await kafkaProducer(params, 'session'); + await sendKafkaMessage(params, 'session'); return getSessionByUuid(session_uuid); } From bb507537046e49b60d63b8975c39064e75187b92 Mon Sep 17 00:00:00 2001 From: Brian Cao Date: Thu, 25 Aug 2022 22:04:32 -0700 Subject: [PATCH 2/8] split db files --- .../migrations/01_init/migration.sql | 6 +- lib/clickhouse.js | 173 +++++++++++ lib/db.js | 286 +----------------- lib/{db => }/kafka.js | 24 +- lib/relational.js | 163 ++++++++++ lib/session.js | 2 + pages/api/collect.js | 39 ++- queries/admin/account/createAccount.js | 2 +- queries/admin/account/deleteAccount.js | 2 +- queries/admin/account/getAccountById.js | 2 +- queries/admin/account/getAccountByUsername.js | 2 +- queries/admin/account/getAccounts.js | 2 +- queries/admin/account/updateAccount.js | 2 +- queries/admin/website/createWebsite.js | 2 +- queries/admin/website/deleteWebsite.js | 2 +- queries/admin/website/getAllWebsites.js | 2 +- queries/admin/website/getUserWebsites.js | 2 +- queries/admin/website/getWebsiteById.js | 2 +- queries/admin/website/getWebsiteByShareId.js | 2 +- queries/admin/website/getWebsiteByUuid.js | 2 +- queries/admin/website/resetWebsite.js | 2 +- queries/admin/website/updateWebsite.js | 2 +- queries/analytics/event/getEventMetrics.js | 20 +- queries/analytics/event/getEvents.js | 14 +- queries/analytics/event/saveEvent.js | 22 +- .../analytics/pageview/getPageviewMetrics.js | 16 +- .../analytics/pageview/getPageviewParams.js | 2 +- .../analytics/pageview/getPageviewStats.js | 23 +- queries/analytics/pageview/getPageviews.js | 14 +- queries/analytics/pageview/savePageView.js | 22 +- queries/analytics/session/createSession.js | 27 +- queries/analytics/session/getSessionByUuid.js | 8 +- .../analytics/session/getSessionMetrics.js | 16 +- queries/analytics/session/getSessions.js | 14 +- queries/analytics/stats/getActiveVisitors.js | 8 +- queries/analytics/stats/getWebsiteStats.js | 21 +- 36 files changed, 496 insertions(+), 454 deletions(-) create mode 100644 lib/clickhouse.js rename lib/{db => }/kafka.js (74%) create mode 100644 lib/relational.js diff --git a/db/clickhouse/migrations/01_init/migration.sql b/db/clickhouse/migrations/01_init/migration.sql index d83ebdab..7794c6fd 100644 --- a/db/clickhouse/migrations/01_init/migration.sql +++ b/db/clickhouse/migrations/01_init/migration.sql @@ -19,7 +19,7 @@ CREATE TABLE pageview_queue ( referrer String ) ENGINE = Kafka -SETTINGS kafka_broker_list = 'localhost:9092,localhost:9093,localhost:9094', -- input broker list +SETTINGS kafka_broker_list = 'dev-01.umami.dev:9092,dev-01.umami.dev:9093,dev-01.umami.dev:9094', -- input broker list kafka_topic_list = 'pageview', kafka_group_name = 'pageview_consumer_group', kafka_format = 'JSONEachRow', @@ -65,7 +65,7 @@ CREATE TABLE session_queue ( country LowCardinality(String) ) ENGINE = Kafka -SETTINGS kafka_broker_list = 'localhost:9092,localhost:9093,localhost:9094', -- input broker list +SETTINGS kafka_broker_list = 'dev-01.umami.dev:9092,dev-01.umami.dev:9093,dev-01.umami.dev:9094', -- input broker list kafka_topic_list = 'session', kafka_group_name = 'session_consumer_group', kafka_format = 'JSONEachRow', @@ -110,7 +110,7 @@ CREATE TABLE event_queue ( event_data String ) ENGINE = Kafka -SETTINGS kafka_broker_list = 'localhost:9092,localhost:9093,localhost:9094', -- input broker list +SETTINGS kafka_broker_list = 'dev-01.umami.dev:9092,dev-01.umami.dev:9093,dev-01.umami.dev:9094', -- input broker list kafka_topic_list = 'event', kafka_group_name = 'event_consumer_group', kafka_format = 'JSONEachRow', diff --git a/lib/clickhouse.js b/lib/clickhouse.js new file mode 100644 index 00000000..3eea9f48 --- /dev/null +++ b/lib/clickhouse.js @@ -0,0 +1,173 @@ +import { ClickHouse } from 'clickhouse'; +import dateFormat from 'dateformat'; +import { FILTER_IGNORED } from 'lib/constants'; +import { CLICKHOUSE_DATE_FORMATS } from './constants'; + +function getClient() { + if (!process.env.ANALYTICS_URL) { + return null; + } + + const url = new URL(process.env.ANALYTICS_URL); + const database = url.pathname.replace('/', ''); + + return new ClickHouse({ + url: url.hostname, + port: Number(url.port), + basicAuth: url.password + ? { + username: url.username || 'default', + password: url.password, + } + : null, + format: 'json', + config: { + database, + }, + }); +} + +const clickhouse = global.clickhouse || getClient(); + +if (process.env.NODE_ENV !== 'production') { + global.clickhouse = clickhouse; +} + +export { clickhouse }; + +export function getDateStringQuery(data, unit) { + return `formatDateTime(${data}, '${CLICKHOUSE_DATE_FORMATS[unit]}')`; +} + +export function getDateQuery(field, unit, timezone) { + if (timezone) { + return `date_trunc('${unit}', ${field}, '${timezone}')`; + } + return `date_trunc('${unit}', ${field})`; +} + +export function getDateFormat(date) { + return `'${dateFormat(date, 'UTC:yyyy-mm-dd HH:MM:ss')}'`; +} + +export function getBetweenDates(field, start_at, end_at) { + return `${field} between ${getDateFormat(start_at)} + and ${getDateFormat(end_at)}`; +} + +export function getFilterQuery(table, column, filters = {}, params = []) { + const query = Object.keys(filters).reduce((arr, key) => { + const filter = filters[key]; + + if (filter === undefined || filter === FILTER_IGNORED) { + return arr; + } + + switch (key) { + case 'url': + if (table === 'pageview' || table === 'event') { + arr.push(`and ${table}.${key}=$${params.length + 1}`); + params.push(decodeURIComponent(filter)); + } + break; + + case 'os': + case 'browser': + case 'device': + case 'country': + if (table === 'session') { + arr.push(`and ${table}.${key}=$${params.length + 1}`); + params.push(decodeURIComponent(filter)); + } + break; + + case 'event_name': + if (table === 'event') { + arr.push(`and ${table}.${key}=$${params.length + 1}`); + params.push(decodeURIComponent(filter)); + } + break; + + case 'referrer': + if (table === 'pageview' || table === 'event') { + arr.push(`and ${table}.referrer like $${params.length + 1}`); + params.push(`%${decodeURIComponent(filter)}%`); + } + break; + + case 'domain': + if (table === 'pageview') { + arr.push(`and ${table}.referrer not like $${params.length + 1}`); + arr.push(`and ${table}.referrer not like '/%'`); + params.push(`%://${filter}/%`); + } + break; + + case 'query': + if (table === 'pageview') { + arr.push(`and ${table}.url like '%?%'`); + } + } + + return arr; + }, []); + + return query.join('\n'); +} + +export function parseFilters(table, column, filters = {}, params = [], sessionKey = 'session_id') { + const { domain, url, event_url, referrer, os, browser, device, country, event_name, query } = + filters; + + const pageviewFilters = { domain, url, referrer, query }; + const sessionFilters = { os, browser, device, country }; + const eventFilters = { url: event_url, event_name }; + + return { + pageviewFilters, + sessionFilters, + eventFilters, + event: { event_name }, + joinSession: + os || browser || device || country + ? `inner join session on ${table}.${sessionKey} = session.${sessionKey}` + : '', + pageviewQuery: getFilterQuery('pageview', column, pageviewFilters, params), + sessionQuery: getFilterQuery('session', column, sessionFilters, params), + eventQuery: getFilterQuery('event', column, eventFilters, params), + }; +} + +export function replaceQuery(string, params = []) { + let formattedString = string; + + params.forEach((a, i) => { + let replace = a; + + if (typeof a === 'string' || a instanceof String) { + replace = `'${replace}'`; + } + + formattedString = formattedString.replace(`$${i + 1}`, replace); + }); + + return formattedString; +} + +export async function rawQuery(query, params = [], debug = false) { + let formattedQuery = replaceQuery(query, params); + + if (debug || process.env.LOG_QUERY) { + console.log(formattedQuery); + } + + return clickhouse.query(formattedQuery).toPromise(); +} + +export async function findUnique(data) { + if (data.length > 1) { + throw `${data.length} records found when expecting 1.`; + } + + return data[0] ?? null; +} diff --git a/lib/db.js b/lib/db.js index 30fa79f6..06443074 100644 --- a/lib/db.js +++ b/lib/db.js @@ -1,305 +1,33 @@ -import { PrismaClient } from '@prisma/client'; -import { ClickHouse } from 'clickhouse'; -import dateFormat from 'dateformat'; -import chalk from 'chalk'; -import { getKafkaService } from './db/kafka'; -import { - MYSQL, - MYSQL_DATE_FORMATS, - POSTGRESQL, - POSTGRESQL_DATE_FORMATS, - CLICKHOUSE, - RELATIONAL, - FILTER_IGNORED, - KAFKA, -} from 'lib/constants'; -import moment from 'moment-timezone'; -import { CLICKHOUSE_DATE_FORMATS } from './constants'; +import { POSTGRESQL, RELATIONAL, MYSQL, KAFKA } from 'lib/constants'; +import { CLICKHOUSE } from 'lib/constants'; BigInt.prototype.toJSON = function () { return Number(this); }; -const options = { - log: [ - { - emit: 'event', - level: 'query', - }, - ], -}; - -function logQuery(e) { - console.log(chalk.yellow(e.params), '->', e.query, chalk.greenBright(`${e.duration}ms`)); -} - -function getPrismaClient(options) { - const prisma = new PrismaClient(options); - - if (process.env.LOG_QUERY) { - prisma.$on('query', logQuery); - } - - return prisma; -} - -function getClickhouseClient() { - if (!process.env.ANALYTICS_URL) { - return null; - } - - const url = new URL(process.env.ANALYTICS_URL); - const database = url.pathname.replace('/', ''); - - return new ClickHouse({ - url: url.hostname, - port: Number(url.port), - basicAuth: url.password - ? { - username: url.username || 'default', - password: url.password, - } - : null, - format: 'json', - config: { - database, - }, - }); -} - -const prisma = global.prisma || getPrismaClient(options); -const clickhouse = global.clickhouse || getClickhouseClient(); - -if (process.env.NODE_ENV !== 'production') { - global.prisma = prisma; - global.clickhouse = clickhouse; -} - -export { prisma, clickhouse }; - -export function getDatabase() { - const type = - process.env.DATABASE_TYPE || - (process.env.DATABASE_URL && process.env.DATABASE_URL.split(':')[0]); - - if (type === 'postgres') { - return POSTGRESQL; - } - - return type; -} - -export function getAnalyticsDatabase() { - const type = process.env.ANALYTICS_URL && process.env.ANALYTICS_URL.split(':')[0]; +export function getDatabase(database, databaseType, fallback) { + const type = databaseType || (database && database.split(':')[0]); if (type === 'postgres') { return POSTGRESQL; } if (!type) { - return getDatabase(); + return getDatabase(fallback); } return type; } -export function getDateStringQueryClickhouse(data, unit) { - return `formatDateTime(${data}, '${CLICKHOUSE_DATE_FORMATS[unit]}')`; -} - -export function getDateQuery(field, unit, timezone) { - const db = getDatabase(); - - if (db === POSTGRESQL) { - if (timezone) { - return `to_char(date_trunc('${unit}', ${field} at time zone '${timezone}'), '${POSTGRESQL_DATE_FORMATS[unit]}')`; - } - return `to_char(date_trunc('${unit}', ${field}), '${POSTGRESQL_DATE_FORMATS[unit]}')`; - } - - if (db === MYSQL) { - if (timezone) { - const tz = moment.tz(timezone).format('Z'); - - return `date_format(convert_tz(${field},'+00:00','${tz}'), '${MYSQL_DATE_FORMATS[unit]}')`; - } - - return `date_format(${field}, '${MYSQL_DATE_FORMATS[unit]}')`; - } -} - -export function getDateQueryClickhouse(field, unit, timezone) { - if (timezone) { - return `date_trunc('${unit}', ${field}, '${timezone}')`; - } - return `date_trunc('${unit}', ${field})`; -} - -export function getDateFormatClickhouse(date) { - return `'${dateFormat(date, 'UTC:yyyy-mm-dd HH:MM:ss')}'`; -} - -export function getBetweenDatesClickhouse(field, start_at, end_at) { - return `${field} between ${getDateFormatClickhouse(start_at)} - and ${getDateFormatClickhouse(end_at)}`; -} - -export function getTimestampInterval(field) { - const db = getDatabase(); - - if (db === POSTGRESQL) { - return `floor(extract(epoch from max(${field}) - min(${field})))`; - } - - if (db === MYSQL) { - return `floor(unix_timestamp(max(${field})) - unix_timestamp(min(${field})))`; - } -} - -export function getFilterQuery(table, column, filters = {}, params = []) { - const query = Object.keys(filters).reduce((arr, key) => { - const filter = filters[key]; - - if (filter === undefined || filter === FILTER_IGNORED) { - return arr; - } - - switch (key) { - case 'url': - if (table === 'pageview' || table === 'event') { - arr.push(`and ${table}.${key}=$${params.length + 1}`); - params.push(decodeURIComponent(filter)); - } - break; - - case 'os': - case 'browser': - case 'device': - case 'country': - if (table === 'session') { - arr.push(`and ${table}.${key}=$${params.length + 1}`); - params.push(decodeURIComponent(filter)); - } - break; - - case 'event_name': - if (table === 'event') { - arr.push(`and ${table}.${key}=$${params.length + 1}`); - params.push(decodeURIComponent(filter)); - } - break; - - case 'referrer': - if (table === 'pageview' || table === 'event') { - arr.push(`and ${table}.referrer like $${params.length + 1}`); - params.push(`%${decodeURIComponent(filter)}%`); - } - break; - - case 'domain': - if (table === 'pageview') { - arr.push(`and ${table}.referrer not like $${params.length + 1}`); - arr.push(`and ${table}.referrer not like '/%'`); - params.push(`%://${filter}/%`); - } - break; - - case 'query': - if (table === 'pageview') { - arr.push(`and ${table}.url like '%?%'`); - } - } - - return arr; - }, []); - - return query.join('\n'); -} - -export function parseFilters(table, column, filters = {}, params = [], sessionKey = 'session_id') { - const { domain, url, event_url, referrer, os, browser, device, country, event_name, query } = - filters; - - const pageviewFilters = { domain, url, referrer, query }; - const sessionFilters = { os, browser, device, country }; - const eventFilters = { url: event_url, event_name }; - - return { - pageviewFilters, - sessionFilters, - eventFilters, - event: { event_name }, - joinSession: - os || browser || device || country - ? `inner join session on ${table}.${sessionKey} = session.${sessionKey}` - : '', - pageviewQuery: getFilterQuery('pageview', column, pageviewFilters, params), - sessionQuery: getFilterQuery('session', column, sessionFilters, params), - eventQuery: getFilterQuery('event', column, eventFilters, params), - }; -} - -export function replaceQueryClickhouse(string, params = []) { - let formattedString = string; - - params.forEach((a, i) => { - let replace = a; - - if (typeof a === 'string' || a instanceof String) { - replace = `'${replace}'`; - } - - formattedString = formattedString.replace(`$${i + 1}`, replace); - }); - - return formattedString; -} - -export async function runQuery(query) { - return query.catch(e => { - throw e; - }); -} - -export async function rawQuery(query, params = []) { - const db = getDatabase(); - - if (db !== POSTGRESQL && db !== MYSQL) { - return Promise.reject(new Error('Unknown database.')); - } - - const sql = db === MYSQL ? query.replace(/\$[0-9]+/g, '?') : query; - - return runQuery(prisma.$queryRawUnsafe.apply(prisma, [sql, ...params])); -} - -export async function rawQueryClickhouse(query, params = [], debug = false) { - let formattedQuery = replaceQueryClickhouse(query, params); - - if (debug || process.env.LOG_QUERY) { - console.log(formattedQuery); - } - - return clickhouse.query(formattedQuery).toPromise(); -} - -export async function findUnique(data) { - if (data.length > 1) { - throw `${data.length} records found when expecting 1.`; - } - - return data[0] ?? null; -} - export async function runAnalyticsQuery(queries) { - const db = getAnalyticsDatabase(); + const db = getDatabase(process.env.ANALYTICS_URL, null, process.env.DATABASE_URL); if (db === POSTGRESQL || db === MYSQL) { return queries[RELATIONAL](); } if (db === CLICKHOUSE) { - const kafka = getKafkaService(); + const kafka = getDatabase(process.env.KAFKA_URL); if (kafka === KAFKA && queries[KAFKA]) { return queries[KAFKA](); } diff --git a/lib/db/kafka.js b/lib/kafka.js similarity index 74% rename from lib/db/kafka.js rename to lib/kafka.js index e2256d27..ab3b3e68 100644 --- a/lib/db/kafka.js +++ b/lib/kafka.js @@ -1,7 +1,7 @@ import { Kafka, logLevel } from 'kafkajs'; import dateFormat from 'dateformat'; -export function getKafkaClient() { +export function getClient() { if (!process.env.KAFKA_URL) { return null; } @@ -30,11 +30,11 @@ export function getKafkaClient() { }); } } -const kafka = global.kafka || getKafkaClient(); +const kafka = global.kafka || getClient(); let kafkaProducer = null; (async () => { - kafkaProducer = global.kakfaProducer || (await getKafkaProducer()); + kafkaProducer = global.kakfaProducer || (await getProducer()); if (process.env.NODE_ENV !== 'production') { global.kafka = kafka; @@ -44,14 +44,18 @@ let kafkaProducer = null; export { kafka, kafkaProducer }; -export async function getKafkaProducer() { +export async function getProducer() { const producer = kafka.producer(); await producer.connect(); return producer; } -export async function sendKafkaMessage(params, topic) { +export function getDateFormat(date) { + return dateFormat(date, 'UTC:yyyy-mm-dd HH:MM:ss'); +} + +export async function sendMessage(params, topic) { await kafkaProducer.send({ topic, messages: [ @@ -63,13 +67,3 @@ export async function sendKafkaMessage(params, topic) { acks: 0, }); } - -export function getDateFormatKafka(date) { - return dateFormat(date, 'UTC:yyyy-mm-dd HH:MM:ss'); -} - -export function getKafkaService() { - const type = process.env.KAFKA_URL && process.env.KAFKA_URL.split(':')[0]; - - return type; -} diff --git a/lib/relational.js b/lib/relational.js new file mode 100644 index 00000000..7d1ed8c2 --- /dev/null +++ b/lib/relational.js @@ -0,0 +1,163 @@ +import { PrismaClient } from '@prisma/client'; +import chalk from 'chalk'; +import { + FILTER_IGNORED, + MYSQL, + MYSQL_DATE_FORMATS, + POSTGRESQL, + POSTGRESQL_DATE_FORMATS, +} from 'lib/constants'; +import { getDatabase } from './db'; +import moment from 'moment-timezone'; + +const options = { + log: [ + { + emit: 'event', + level: 'query', + }, + ], +}; + +function logQuery(e) { + console.log(chalk.yellow(e.params), '->', e.query, chalk.greenBright(`${e.duration}ms`)); +} + +function getPrismaClient(options) { + const prisma = new PrismaClient(options); + + if (process.env.LOG_QUERY) { + prisma.$on('query', logQuery); + } + + return prisma; +} +const prisma = global.prisma || getPrismaClient(options); + +if (process.env.NODE_ENV !== 'production') { + global.prisma = prisma; +} + +export { prisma }; + +export function getDateQuery(field, unit, timezone) { + const db = getDatabase(); + + if (db === POSTGRESQL) { + if (timezone) { + return `to_char(date_trunc('${unit}', ${field} at time zone '${timezone}'), '${POSTGRESQL_DATE_FORMATS[unit]}')`; + } + return `to_char(date_trunc('${unit}', ${field}), '${POSTGRESQL_DATE_FORMATS[unit]}')`; + } + + if (db === MYSQL) { + if (timezone) { + const tz = moment.tz(timezone).format('Z'); + + return `date_format(convert_tz(${field},'+00:00','${tz}'), '${MYSQL_DATE_FORMATS[unit]}')`; + } + + return `date_format(${field}, '${MYSQL_DATE_FORMATS[unit]}')`; + } +} + +export function getFilterQuery(table, column, filters = {}, params = []) { + const query = Object.keys(filters).reduce((arr, key) => { + const filter = filters[key]; + + if (filter === undefined || filter === FILTER_IGNORED) { + return arr; + } + + switch (key) { + case 'url': + if (table === 'pageview' || table === 'event') { + arr.push(`and ${table}.${key}=$${params.length + 1}`); + params.push(decodeURIComponent(filter)); + } + break; + + case 'os': + case 'browser': + case 'device': + case 'country': + if (table === 'session') { + arr.push(`and ${table}.${key}=$${params.length + 1}`); + params.push(decodeURIComponent(filter)); + } + break; + + case 'event_name': + if (table === 'event') { + arr.push(`and ${table}.${key}=$${params.length + 1}`); + params.push(decodeURIComponent(filter)); + } + break; + + case 'referrer': + if (table === 'pageview' || table === 'event') { + arr.push(`and ${table}.referrer like $${params.length + 1}`); + params.push(`%${decodeURIComponent(filter)}%`); + } + break; + + case 'domain': + if (table === 'pageview') { + arr.push(`and ${table}.referrer not like $${params.length + 1}`); + arr.push(`and ${table}.referrer not like '/%'`); + params.push(`%://${filter}/%`); + } + break; + + case 'query': + if (table === 'pageview') { + arr.push(`and ${table}.url like '%?%'`); + } + } + + return arr; + }, []); + + return query.join('\n'); +} + +export function parseFilters(table, column, filters = {}, params = [], sessionKey = 'session_id') { + const { domain, url, event_url, referrer, os, browser, device, country, event_name, query } = + filters; + + const pageviewFilters = { domain, url, referrer, query }; + const sessionFilters = { os, browser, device, country }; + const eventFilters = { url: event_url, event_name }; + + return { + pageviewFilters, + sessionFilters, + eventFilters, + event: { event_name }, + joinSession: + os || browser || device || country + ? `inner join session on ${table}.${sessionKey} = session.${sessionKey}` + : '', + pageviewQuery: getFilterQuery('pageview', column, pageviewFilters, params), + sessionQuery: getFilterQuery('session', column, sessionFilters, params), + eventQuery: getFilterQuery('event', column, eventFilters, params), + }; +} + +export async function runQuery(query) { + return query.catch(e => { + throw e; + }); +} + +export async function rawQuery(query, params = []) { + const db = getDatabase(process.env.DATABASE_URL, process.env.DATABASE_TYPE); + + if (db !== POSTGRESQL && db !== MYSQL) { + return Promise.reject(new Error('Unknown database.')); + } + + const sql = db === MYSQL ? query.replace(/\$[0-9]+/g, '?') : query; + + return runQuery(prisma.$queryRawUnsafe.apply(prisma, [sql, ...params])); +} diff --git a/lib/session.js b/lib/session.js index c9fb04d7..b90e88f7 100644 --- a/lib/session.js +++ b/lib/session.js @@ -37,6 +37,8 @@ export async function getSession(req) { let session = await getSessionByUuid(session_uuid); + console.log('session here!: ', session); + if (!session) { try { session = await createSession(website_id, { diff --git a/pages/api/collect.js b/pages/api/collect.js index 1cff4841..4b976319 100644 --- a/pages/api/collect.js +++ b/pages/api/collect.js @@ -4,7 +4,7 @@ import ipaddr from 'ipaddr.js'; import { savePageView, saveEvent } from 'queries'; import { useCors, useSession } from 'lib/middleware'; import { getJsonBody, getIpAddress } from 'lib/request'; -import { ok, send, badRequest, forbidden } from 'lib/response'; +import { unauthorized, send, badRequest, forbidden } from 'lib/response'; import { createToken } from 'lib/crypto'; import { removeTrailingSlash } from 'lib/url'; import { uuid } from 'lib/crypto'; @@ -13,7 +13,7 @@ export default async (req, res) => { await useCors(req, res); if (isbot(req.headers['user-agent'])) { - return ok(res); + return unauthorized(res); } const ignoreIps = process.env.IGNORE_IP; @@ -93,3 +93,38 @@ export default async (req, res) => { return send(res, token); }; + +// async function relational(req, res) { +// await useSession(req, res); + +// const { +// session: { website_id, session_id, session_uuid }, +// } = req; + +// const { type, payload } = getJsonBody(req); + +// let { url, referrer, event_name, event_data } = payload; + +// if (process.env.REMOVE_TRAILING_SLASH) { +// url = removeTrailingSlash(url); +// } + +// const event_uuid = uuid(); + +// if (type === 'pageview') { +// await savePageView(website_id, { session_id, session_uuid, url, referrer }); +// } else if (type === 'event') { +// await saveEvent(website_id, { +// event_uuid, +// session_id, +// session_uuid, +// url, +// event_name, +// event_data, +// }); +// } else { +// return badRequest(res); +// } +// } + +// function clickhouse() {} diff --git a/queries/admin/account/createAccount.js b/queries/admin/account/createAccount.js index 193ed719..928f9f45 100644 --- a/queries/admin/account/createAccount.js +++ b/queries/admin/account/createAccount.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function createAccount(data) { return runQuery( diff --git a/queries/admin/account/deleteAccount.js b/queries/admin/account/deleteAccount.js index 8c4e2c0c..6a5896d4 100644 --- a/queries/admin/account/deleteAccount.js +++ b/queries/admin/account/deleteAccount.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function deleteAccount(user_id) { return runQuery( diff --git a/queries/admin/account/getAccountById.js b/queries/admin/account/getAccountById.js index de2605e1..92fe5b43 100644 --- a/queries/admin/account/getAccountById.js +++ b/queries/admin/account/getAccountById.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function getAccountById(user_id) { return runQuery( diff --git a/queries/admin/account/getAccountByUsername.js b/queries/admin/account/getAccountByUsername.js index d476c38e..a777c0ed 100644 --- a/queries/admin/account/getAccountByUsername.js +++ b/queries/admin/account/getAccountByUsername.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function getAccountByUsername(username) { return runQuery( diff --git a/queries/admin/account/getAccounts.js b/queries/admin/account/getAccounts.js index e6ebedf2..e24476df 100644 --- a/queries/admin/account/getAccounts.js +++ b/queries/admin/account/getAccounts.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function getAccounts() { return runQuery( diff --git a/queries/admin/account/updateAccount.js b/queries/admin/account/updateAccount.js index b8d718f4..2137003f 100644 --- a/queries/admin/account/updateAccount.js +++ b/queries/admin/account/updateAccount.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function updateAccount(user_id, data) { return runQuery( diff --git a/queries/admin/website/createWebsite.js b/queries/admin/website/createWebsite.js index 3cb6887f..8770d1ce 100644 --- a/queries/admin/website/createWebsite.js +++ b/queries/admin/website/createWebsite.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function createWebsite(user_id, data) { return runQuery( diff --git a/queries/admin/website/deleteWebsite.js b/queries/admin/website/deleteWebsite.js index 129351ea..5d88a3b5 100644 --- a/queries/admin/website/deleteWebsite.js +++ b/queries/admin/website/deleteWebsite.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function deleteWebsite(website_id) { return runQuery( diff --git a/queries/admin/website/getAllWebsites.js b/queries/admin/website/getAllWebsites.js index 7e413fe8..03e7d86e 100644 --- a/queries/admin/website/getAllWebsites.js +++ b/queries/admin/website/getAllWebsites.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function getAllWebsites() { let data = await runQuery( diff --git a/queries/admin/website/getUserWebsites.js b/queries/admin/website/getUserWebsites.js index 4cd75997..4db70aab 100644 --- a/queries/admin/website/getUserWebsites.js +++ b/queries/admin/website/getUserWebsites.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function getUserWebsites(user_id) { return runQuery( diff --git a/queries/admin/website/getWebsiteById.js b/queries/admin/website/getWebsiteById.js index 8cf0f50f..231cd899 100644 --- a/queries/admin/website/getWebsiteById.js +++ b/queries/admin/website/getWebsiteById.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function getWebsiteById(website_id) { return runQuery( diff --git a/queries/admin/website/getWebsiteByShareId.js b/queries/admin/website/getWebsiteByShareId.js index 9fcc1ab3..3148b062 100644 --- a/queries/admin/website/getWebsiteByShareId.js +++ b/queries/admin/website/getWebsiteByShareId.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function getWebsiteByShareId(share_id) { return runQuery( diff --git a/queries/admin/website/getWebsiteByUuid.js b/queries/admin/website/getWebsiteByUuid.js index 36c41e67..29e48a06 100644 --- a/queries/admin/website/getWebsiteByUuid.js +++ b/queries/admin/website/getWebsiteByUuid.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function getWebsiteByUuid(website_uuid) { return runQuery( diff --git a/queries/admin/website/resetWebsite.js b/queries/admin/website/resetWebsite.js index 18f249b8..8a1c426c 100644 --- a/queries/admin/website/resetWebsite.js +++ b/queries/admin/website/resetWebsite.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function resetWebsite(website_id) { return runQuery(prisma.$queryRaw`delete from session where website_id=${website_id}`); diff --git a/queries/admin/website/updateWebsite.js b/queries/admin/website/updateWebsite.js index e2a16925..189bd938 100644 --- a/queries/admin/website/updateWebsite.js +++ b/queries/admin/website/updateWebsite.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function updateWebsite(website_id, data) { return runQuery( diff --git a/queries/analytics/event/getEventMetrics.js b/queries/analytics/event/getEventMetrics.js index 8a624f92..41111e71 100644 --- a/queries/analytics/event/getEventMetrics.js +++ b/queries/analytics/event/getEventMetrics.js @@ -1,13 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { - getBetweenDatesClickhouse, - getDateQuery, - getDateQueryClickhouse, - getFilterQuery, - rawQuery, - rawQueryClickhouse, - runAnalyticsQuery, -} from 'lib/db'; +import clickhouse from 'lib/clickhouse'; +import { getDateQuery, getFilterQuery, rawQuery } from 'lib/db/relational'; +import { runAnalyticsQuery } from 'lib/db/db'; export async function getEventMetrics(...args) { return runAnalyticsQuery({ @@ -53,16 +47,16 @@ async function clickhouseQuery( ) { const params = [website_id]; - return rawQueryClickhouse( + return clickhouse.rawQuery( ` select event_name x, - ${getDateQueryClickhouse('created_at', unit, timezone)} t, + ${clickhouse.getDateQuery('created_at', unit, timezone)} t, count(*) y from event where website_id= $1 - and ${getBetweenDatesClickhouse('created_at', start_at, end_at)} - ${getFilterQuery('event', filters, params)} + and ${clickhouse.getBetweenDates('created_at', start_at, end_at)} + ${clickhouse.getFilterQuery('event', filters, params)} group by x, t order by t `, diff --git a/queries/analytics/event/getEvents.js b/queries/analytics/event/getEvents.js index 7969877a..42016fc0 100644 --- a/queries/analytics/event/getEvents.js +++ b/queries/analytics/event/getEvents.js @@ -1,11 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { - rawQueryClickhouse, - getDateFormatClickhouse, - prisma, - runAnalyticsQuery, - runQuery, -} from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; +import clickhouse from 'lib/clickhouse'; +import { runAnalyticsQuery } from 'lib/db/db'; export function getEvents(...args) { return runAnalyticsQuery({ @@ -32,7 +28,7 @@ function relationalQuery(websites, start_at) { } function clickhouseQuery(websites, start_at) { - return rawQueryClickhouse( + return clickhouse.rawQuery( ` select event_id, @@ -43,7 +39,7 @@ function clickhouseQuery(websites, start_at) { event_name from event where website_id in (${websites.join[',']} - and created_at >= ${getDateFormatClickhouse(start_at)}) + and created_at >= ${clickhouse.getDateFormat(start_at)}) `, ); } diff --git a/queries/analytics/event/saveEvent.js b/queries/analytics/event/saveEvent.js index ff4d5256..fc73c8b8 100644 --- a/queries/analytics/event/saveEvent.js +++ b/queries/analytics/event/saveEvent.js @@ -1,12 +1,8 @@ -import { CLICKHOUSE, RELATIONAL, KAFKA, URL_LENGTH } from 'lib/constants'; -import { - getDateFormatClickhouse, - prisma, - rawQueryClickhouse, - runAnalyticsQuery, - runQuery, -} from 'lib/db'; -import { sendKafkaMessage, getDateFormatKafka } from 'lib/db/kafka'; +import { CLICKHOUSE, KAFKA, RELATIONAL, URL_LENGTH } from 'lib/constants'; +import clickhouse from 'lib/clickhouse'; +import kafka from 'lib/db/kafka'; +import { prisma, runQuery } from 'lib/db/relational'; +import { runAnalyticsQuery } from 'lib/db/db'; export async function saveEvent(...args) { return runAnalyticsQuery({ @@ -48,10 +44,10 @@ async function clickhouseQuery(website_id, { event_uuid, session_uuid, url, even event_name?.substr(0, 50), ]; - return rawQueryClickhouse( + return clickhouse.rawQuery( ` insert into umami.event (created_at, website_id, session_uuid, url, event_name) - values (${getDateFormatClickhouse(new Date())}, $1, $2, $3, $4);`, + values (${clickhouse.getDateFormat(new Date())}, $1, $2, $3, $4);`, params, ); } @@ -61,10 +57,10 @@ async function kafkaQuery(website_id, { event_uuid, session_uuid, url, event_nam event_uuid: event_uuid, website_id: website_id, session_uuid: session_uuid, - created_at: getDateFormatKafka(new Date()), + created_at: kafka.getDateFormat(new Date()), url: url?.substr(0, URL_LENGTH), event_name: event_name?.substr(0, 50), }; - await sendKafkaMessage(params, 'event'); + await kafka.sendKafkaMessage(params, 'event'); } diff --git a/queries/analytics/pageview/getPageviewMetrics.js b/queries/analytics/pageview/getPageviewMetrics.js index 9d3eabaa..3997fd19 100644 --- a/queries/analytics/pageview/getPageviewMetrics.js +++ b/queries/analytics/pageview/getPageviewMetrics.js @@ -1,11 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { - rawQueryClickhouse, - runAnalyticsQuery, - parseFilters, - rawQuery, - getBetweenDatesClickhouse, -} from 'lib/db'; +import clickhouse from 'lib/clickhouse'; +import { parseFilters, rawQuery } from 'lib/db/relational'; +import { runAnalyticsQuery } from 'lib/db/db'; export async function getPageviewMetrics(...args) { return runAnalyticsQuery({ @@ -42,7 +38,7 @@ async function relationalQuery(website_id, start_at, end_at, column, table, filt async function clickhouseQuery(website_id, start_at, end_at, column, table, filters = {}) { const params = [website_id]; - const { pageviewQuery, sessionQuery, eventQuery, joinSession } = parseFilters( + const { pageviewQuery, sessionQuery, eventQuery, joinSession } = clickhouse.parseFilters( table, column, filters, @@ -50,13 +46,13 @@ async function clickhouseQuery(website_id, start_at, end_at, column, table, filt 'session_uuid', ); - return rawQueryClickhouse( + return clickhouse.rawQuery( ` select ${column} x, count(*) y from ${table} ${joinSession} where ${table}.website_id= $1 - and ${getBetweenDatesClickhouse(table + '.created_at', start_at, end_at)} + and ${clickhouse.getBetweenDates(table + '.created_at', start_at, end_at)} ${pageviewQuery} ${joinSession && sessionQuery} ${eventQuery} diff --git a/queries/analytics/pageview/getPageviewParams.js b/queries/analytics/pageview/getPageviewParams.js index 549ff11c..d77d328e 100644 --- a/queries/analytics/pageview/getPageviewParams.js +++ b/queries/analytics/pageview/getPageviewParams.js @@ -1,4 +1,4 @@ -import { parseFilters, rawQuery, runAnalyticsQuery } from 'lib/db'; +import { parseFilters, rawQuery, runAnalyticsQuery } from 'lib/db/relational'; import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; export async function getPageviewParams(...args) { diff --git a/queries/analytics/pageview/getPageviewStats.js b/queries/analytics/pageview/getPageviewStats.js index 00ff857a..6a786f58 100644 --- a/queries/analytics/pageview/getPageviewStats.js +++ b/queries/analytics/pageview/getPageviewStats.js @@ -1,14 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { - getBetweenDatesClickhouse, - getDateQuery, - getDateQueryClickhouse, - getDateStringQueryClickhouse, - parseFilters, - rawQuery, - rawQueryClickhouse, - runAnalyticsQuery, -} from 'lib/db'; +import { getDateQuery, parseFilters, rawQuery } from 'lib/db/relational'; +import { runAnalyticsQuery } from 'lib/db/db'; +import clickhouse from 'lib/clickhouse'; export async function getPageviewStats(...args) { return runAnalyticsQuery({ @@ -62,7 +55,7 @@ async function clickhouseQuery( sessionKey = 'session_uuid', ) { const params = [website_id]; - const { pageviewQuery, sessionQuery, joinSession } = parseFilters( + const { pageviewQuery, sessionQuery, joinSession } = clickhouse.parseFilters( 'pageview', null, filters, @@ -70,19 +63,19 @@ async function clickhouseQuery( sessionKey, ); - return rawQueryClickhouse( + return clickhouse.rawQuery( ` select - ${getDateStringQueryClickhouse('g.t', unit)} as t, + ${clickhouse.getDateStringQuery('g.t', unit)} as t, g.y as y from (select - ${getDateQueryClickhouse('created_at', unit, timezone)} t, + ${clickhouse.getDateQuery('created_at', unit, timezone)} t, count(${count !== '*' ? `${count}${sessionKey}` : count}) y from pageview ${joinSession} where pageview.website_id= $1 - and ${getBetweenDatesClickhouse('pageview.created_at', start_at, end_at)} + and ${clickhouse.getBetweenDates('pageview.created_at', start_at, end_at)} ${pageviewQuery} ${sessionQuery} group by t) g diff --git a/queries/analytics/pageview/getPageviews.js b/queries/analytics/pageview/getPageviews.js index 45463ec8..bccc5bdb 100644 --- a/queries/analytics/pageview/getPageviews.js +++ b/queries/analytics/pageview/getPageviews.js @@ -1,11 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { - rawQueryClickhouse, - getDateFormatClickhouse, - prisma, - runAnalyticsQuery, - runQuery, -} from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; +import { runAnalyticsQuery } from 'lib/db/db'; +import clickhouse from 'lib/clickhouse'; export async function getPageviews(...args) { return runAnalyticsQuery({ @@ -32,7 +28,7 @@ async function relationalQuery(websites, start_at) { } async function clickhouseQuery(websites, start_at) { - return rawQueryClickhouse( + return clickhouse.rawQuery( ` select view_id, @@ -42,7 +38,7 @@ async function clickhouseQuery(websites, start_at) { url from pageview where website_id in (${websites.join[',']} - and created_at >= ${getDateFormatClickhouse(start_at)}) + and created_at >= ${clickhouse.getDateFormat(start_at)}) `, ); } diff --git a/queries/analytics/pageview/savePageView.js b/queries/analytics/pageview/savePageView.js index af0d88f6..926b0ad8 100644 --- a/queries/analytics/pageview/savePageView.js +++ b/queries/analytics/pageview/savePageView.js @@ -1,12 +1,8 @@ -import { CLICKHOUSE, RELATIONAL, KAFKA, URL_LENGTH } from 'lib/constants'; -import { - getDateFormatClickhouse, - prisma, - rawQueryClickhouse, - runAnalyticsQuery, - runQuery, -} from 'lib/db'; -import { sendKafkaMessage, getDateFormatKafka } from 'lib/db/kafka'; +import { CLICKHOUSE, KAFKA, RELATIONAL, URL_LENGTH } from 'lib/constants'; +import clickhouse from 'lib/clickhouse'; +import { runAnalyticsQuery } from 'lib/db/db'; +import kafka from 'lib/db/kafka'; +import { prisma, runQuery } from 'lib/db/relational'; export async function savePageView(...args) { return runAnalyticsQuery({ @@ -37,10 +33,10 @@ async function clickhouseQuery(website_id, { session_uuid, url, referrer }) { referrer?.substr(0, URL_LENGTH), ]; - return rawQueryClickhouse( + return clickhouse.rawQuery( ` insert into umami.pageview (created_at, website_id, session_uuid, url, referrer) - values (${getDateFormatClickhouse(new Date())}, $1, $2, $3, $4);`, + values (${clickhouse.getDateFormat(new Date())}, $1, $2, $3, $4);`, params, ); } @@ -49,10 +45,10 @@ async function kafkaQuery(website_id, { session_uuid, url, referrer }) { const params = { website_id: website_id, session_uuid: session_uuid, - created_at: getDateFormatKafka(new Date()), + created_at: kafka.getDateFormat(new Date()), url: url?.substr(0, URL_LENGTH), referrer: referrer?.substr(0, URL_LENGTH), }; - await sendKafkaMessage(params, 'pageview'); + await kafka.sendKafkaMessage(params, 'pageview'); } diff --git a/queries/analytics/session/createSession.js b/queries/analytics/session/createSession.js index 565d45a2..b6dcd563 100644 --- a/queries/analytics/session/createSession.js +++ b/queries/analytics/session/createSession.js @@ -1,13 +1,8 @@ -import { CLICKHOUSE, RELATIONAL, KAFKA } from 'lib/constants'; -import { - getDateFormatClickhouse, - prisma, - rawQueryClickhouse, - runAnalyticsQuery, - runQuery, -} from 'lib/db'; -import { sendKafkaMessage, getDateFormatKafka } from 'lib/db/kafka'; -import { getSessionByUuid } from 'queries'; +import { CLICKHOUSE, KAFKA, RELATIONAL } from 'lib/constants'; +import { prisma, runQuery } from 'lib/db/relational'; +import clickhouse from 'lib/clickhouse'; +import kafka from 'lib/db/kafka'; +import { runAnalyticsQuery } from 'lib/db/db'; export async function createSession(...args) { return runAnalyticsQuery({ @@ -47,13 +42,11 @@ async function clickhouseQuery( country ? country : null, ]; - await rawQueryClickhouse( + await clickhouse.rawQuery( `insert into umami.session (created_at, session_uuid, website_id, hostname, browser, os, device, screen, language, country) - values (${getDateFormatClickhouse(new Date())}, $1, $2, $3, $4, $5, $6, $7, $8, $9);`, + values (${clickhouse.getDateFormat(new Date())}, $1, $2, $3, $4, $5, $6, $7, $8, $9);`, params, ); - - return getSessionByUuid(session_uuid); } async function kafkaQuery( @@ -63,7 +56,7 @@ async function kafkaQuery( const params = { session_uuid: session_uuid, website_id: website_id, - created_at: getDateFormatKafka(new Date()), + created_at: kafka.getDateFormat(new Date()), hostname: hostname, browser: browser, os: os, @@ -73,7 +66,5 @@ async function kafkaQuery( country: country ? country : null, }; - await sendKafkaMessage(params, 'session'); - - return getSessionByUuid(session_uuid); + await kafka.sendKafkaMessage(params, 'session'); } diff --git a/queries/analytics/session/getSessionByUuid.js b/queries/analytics/session/getSessionByUuid.js index 2b9acc1e..6fa6dfba 100644 --- a/queries/analytics/session/getSessionByUuid.js +++ b/queries/analytics/session/getSessionByUuid.js @@ -1,5 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { rawQueryClickhouse, findUnique, prisma, runAnalyticsQuery, runQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/db/relational'; +import clickhouse from 'lib/clickhouse'; +import { runAnalyticsQuery } from 'lib/db/db'; export async function getSessionByUuid(...args) { return runAnalyticsQuery({ @@ -21,7 +23,7 @@ async function relationalQuery(session_uuid) { async function clickhouseQuery(session_uuid) { const params = [session_uuid]; - return rawQueryClickhouse( + return clickhouse.rawQuery( ` select session_uuid, @@ -38,5 +40,5 @@ async function clickhouseQuery(session_uuid) { where session_uuid = $1 `, params, - ).then(data => findUnique(data)); + ); } diff --git a/queries/analytics/session/getSessionMetrics.js b/queries/analytics/session/getSessionMetrics.js index 7da60bda..6fde5b3d 100644 --- a/queries/analytics/session/getSessionMetrics.js +++ b/queries/analytics/session/getSessionMetrics.js @@ -1,11 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { - getBetweenDatesClickhouse, - parseFilters, - rawQuery, - rawQueryClickhouse, - runAnalyticsQuery, -} from 'lib/db'; +import clickhouse from 'lib/clickhouse'; +import { runAnalyticsQuery } from 'lib/db/db'; +import { parseFilters, rawQuery } from 'lib/db/relational'; export async function getSessionMetrics(...args) { return runAnalyticsQuery({ @@ -45,7 +41,7 @@ async function relationalQuery(website_id, start_at, end_at, field, filters = {} async function clickhouseQuery(website_id, start_at, end_at, field, filters = {}) { const params = [website_id]; - const { pageviewQuery, sessionQuery, joinSession } = parseFilters( + const { pageviewQuery, sessionQuery, joinSession } = clickhouse.parseFilters( 'pageview', null, filters, @@ -53,7 +49,7 @@ async function clickhouseQuery(website_id, start_at, end_at, field, filters = {} 'session_uuid', ); - return rawQueryClickhouse( + return clickhouse.rawQuery( ` select ${field} x, count(*) y from session as x @@ -62,7 +58,7 @@ async function clickhouseQuery(website_id, start_at, end_at, field, filters = {} from pageview ${joinSession} where pageview.website_id=$1 - and ${getBetweenDatesClickhouse('pageview.created_at', start_at, end_at)} + and ${clickhouse.getBetweenDates('pageview.created_at', start_at, end_at)} ${pageviewQuery} ${sessionQuery} ) diff --git a/queries/analytics/session/getSessions.js b/queries/analytics/session/getSessions.js index bc48db1c..57806c6f 100644 --- a/queries/analytics/session/getSessions.js +++ b/queries/analytics/session/getSessions.js @@ -1,11 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { - getDateFormatClickhouse, - prisma, - rawQueryClickhouse, - runAnalyticsQuery, - runQuery, -} from 'lib/db'; +import clickhouse from 'lib/clickhouse'; +import { runAnalyticsQuery } from 'lib/db/db'; +import { prisma, runQuery } from 'lib/db/relational'; export async function getSessions(...args) { return runAnalyticsQuery({ @@ -32,7 +28,7 @@ async function relationalQuery(websites, start_at) { } async function clickhouseQuery(websites, start_at) { - return rawQueryClickhouse( + return clickhouse.rawQuery( ` select session_id, @@ -48,7 +44,7 @@ async function clickhouseQuery(websites, start_at) { country from session where website_id in (${websites.join[',']} - and created_at >= ${getDateFormatClickhouse(start_at)}) + and created_at >= ${clickhouse.getDateFormat(start_at)}) `, ); } diff --git a/queries/analytics/stats/getActiveVisitors.js b/queries/analytics/stats/getActiveVisitors.js index aac99085..fa525314 100644 --- a/queries/analytics/stats/getActiveVisitors.js +++ b/queries/analytics/stats/getActiveVisitors.js @@ -1,6 +1,8 @@ import { subMinutes } from 'date-fns'; import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { getDateFormatClickhouse, rawQuery, rawQueryClickhouse, runAnalyticsQuery } from 'lib/db'; +import { rawQuery } from 'lib/db/relational'; +import { runAnalyticsQuery } from 'lib/db/db'; +import clickhouse from 'lib/clickhouse'; export async function getActiveVisitors(...args) { return runAnalyticsQuery({ @@ -27,12 +29,12 @@ async function relationalQuery(website_id) { async function clickhouseQuery(website_id) { const params = [website_id]; - return rawQueryClickhouse( + return clickhouse.rawQuery( ` select count(distinct session_uuid) x from pageview where website_id = $1 - and created_at >= ${getDateFormatClickhouse(subMinutes(new Date(), 5))} + and created_at >= ${clickhouse.getDateFormat(subMinutes(new Date(), 5))} `, params, ); diff --git a/queries/analytics/stats/getWebsiteStats.js b/queries/analytics/stats/getWebsiteStats.js index dc0380e0..e4fb4eb8 100644 --- a/queries/analytics/stats/getWebsiteStats.js +++ b/queries/analytics/stats/getWebsiteStats.js @@ -1,14 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { - getDateQuery, - getBetweenDatesClickhouse, - getDateQueryClickhouse, - getTimestampInterval, - parseFilters, - rawQuery, - rawQueryClickhouse, - runAnalyticsQuery, -} from 'lib/db'; +import { getDateQuery, getTimestampInterval, parseFilters, rawQuery } from 'lib/db/relational'; +import { runAnalyticsQuery } from 'lib/db/db'; +import clickhouse from 'lib/clickhouse'; export async function getWebsiteStats(...args) { return runAnalyticsQuery({ @@ -52,7 +45,7 @@ async function relationalQuery(website_id, start_at, end_at, filters = {}) { async function clickhouseQuery(website_id, start_at, end_at, filters = {}) { const params = [website_id]; - const { pageviewQuery, sessionQuery, joinSession } = parseFilters( + const { pageviewQuery, sessionQuery, joinSession } = clickhouse.parseFilters( 'pageview', null, filters, @@ -60,7 +53,7 @@ async function clickhouseQuery(website_id, start_at, end_at, filters = {}) { 'session_uuid', ); - return rawQueryClickhouse( + return clickhouse.rawQuery( ` select sum(t.c) as "pageviews", @@ -69,14 +62,14 @@ async function clickhouseQuery(website_id, start_at, end_at, filters = {}) { sum(if(max_time < min_time + interval 1 hour, max_time-min_time, 0)) as "totaltime" from ( select pageview.session_uuid, - ${getDateQueryClickhouse('pageview.created_at', 'day')} time_series, + ${clickhouse.getDateQuery('pageview.created_at', 'day')} time_series, count(*) c, min(created_at) min_time, max(created_at) max_time from pageview ${joinSession} where pageview.website_id = $1 - and ${getBetweenDatesClickhouse('pageview.created_at', start_at, end_at)} + and ${clickhouse.getBetweenDates('pageview.created_at', start_at, end_at)} ${pageviewQuery} ${sessionQuery} group by pageview.session_uuid, time_series From f69bd36223b797d555e50b038326a75500c124bb Mon Sep 17 00:00:00 2001 From: Brian Cao Date: Thu, 25 Aug 2022 22:05:34 -0700 Subject: [PATCH 3/8] fix migration --- db/clickhouse/migrations/01_init/migration.sql | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/db/clickhouse/migrations/01_init/migration.sql b/db/clickhouse/migrations/01_init/migration.sql index 7794c6fd..d83ebdab 100644 --- a/db/clickhouse/migrations/01_init/migration.sql +++ b/db/clickhouse/migrations/01_init/migration.sql @@ -19,7 +19,7 @@ CREATE TABLE pageview_queue ( referrer String ) ENGINE = Kafka -SETTINGS kafka_broker_list = 'dev-01.umami.dev:9092,dev-01.umami.dev:9093,dev-01.umami.dev:9094', -- input broker list +SETTINGS kafka_broker_list = 'localhost:9092,localhost:9093,localhost:9094', -- input broker list kafka_topic_list = 'pageview', kafka_group_name = 'pageview_consumer_group', kafka_format = 'JSONEachRow', @@ -65,7 +65,7 @@ CREATE TABLE session_queue ( country LowCardinality(String) ) ENGINE = Kafka -SETTINGS kafka_broker_list = 'dev-01.umami.dev:9092,dev-01.umami.dev:9093,dev-01.umami.dev:9094', -- input broker list +SETTINGS kafka_broker_list = 'localhost:9092,localhost:9093,localhost:9094', -- input broker list kafka_topic_list = 'session', kafka_group_name = 'session_consumer_group', kafka_format = 'JSONEachRow', @@ -110,7 +110,7 @@ CREATE TABLE event_queue ( event_data String ) ENGINE = Kafka -SETTINGS kafka_broker_list = 'dev-01.umami.dev:9092,dev-01.umami.dev:9093,dev-01.umami.dev:9094', -- input broker list +SETTINGS kafka_broker_list = 'localhost:9092,localhost:9093,localhost:9094', -- input broker list kafka_topic_list = 'event', kafka_group_name = 'event_consumer_group', kafka_format = 'JSONEachRow', From c811577460d0450adcd5ab236ac60520e0c35e05 Mon Sep 17 00:00:00 2001 From: Brian Cao Date: Thu, 25 Aug 2022 22:20:30 -0700 Subject: [PATCH 4/8] fix folders --- lib/clickhouse.js | 4 ++++ queries/admin/account/createAccount.js | 2 +- queries/admin/account/deleteAccount.js | 2 +- queries/admin/account/getAccountById.js | 2 +- queries/admin/account/getAccountByUsername.js | 2 +- queries/admin/account/getAccounts.js | 2 +- queries/admin/account/updateAccount.js | 2 +- queries/admin/website/createWebsite.js | 2 +- queries/admin/website/deleteWebsite.js | 2 +- queries/admin/website/getAllWebsites.js | 2 +- queries/admin/website/getUserWebsites.js | 2 +- queries/admin/website/getWebsiteById.js | 2 +- queries/admin/website/getWebsiteByShareId.js | 2 +- queries/admin/website/getWebsiteByUuid.js | 2 +- queries/admin/website/resetWebsite.js | 2 +- queries/admin/website/updateWebsite.js | 2 +- queries/analytics/event/getEventMetrics.js | 2 +- queries/analytics/event/getEvents.js | 2 +- queries/analytics/event/saveEvent.js | 4 ++-- queries/analytics/pageview/getPageviewMetrics.js | 2 +- queries/analytics/pageview/getPageviewParams.js | 2 +- queries/analytics/pageview/getPageviewStats.js | 2 +- queries/analytics/pageview/getPageviews.js | 2 +- queries/analytics/pageview/savePageView.js | 2 +- queries/analytics/session/createSession.js | 2 +- queries/analytics/session/getSessionByUuid.js | 14 ++++++++------ queries/analytics/session/getSessionMetrics.js | 2 +- queries/analytics/session/getSessions.js | 4 ++-- queries/analytics/stats/getActiveVisitors.js | 2 +- queries/analytics/stats/getWebsiteStats.js | 2 +- 30 files changed, 42 insertions(+), 36 deletions(-) diff --git a/lib/clickhouse.js b/lib/clickhouse.js index 3eea9f48..ea85d579 100644 --- a/lib/clickhouse.js +++ b/lib/clickhouse.js @@ -171,3 +171,7 @@ export async function findUnique(data) { return data[0] ?? null; } + +export async function findFirst(data) { + return data[0] ?? null; +} diff --git a/queries/admin/account/createAccount.js b/queries/admin/account/createAccount.js index 928f9f45..fb3997e8 100644 --- a/queries/admin/account/createAccount.js +++ b/queries/admin/account/createAccount.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function createAccount(data) { return runQuery( diff --git a/queries/admin/account/deleteAccount.js b/queries/admin/account/deleteAccount.js index 6a5896d4..f03b370e 100644 --- a/queries/admin/account/deleteAccount.js +++ b/queries/admin/account/deleteAccount.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function deleteAccount(user_id) { return runQuery( diff --git a/queries/admin/account/getAccountById.js b/queries/admin/account/getAccountById.js index 92fe5b43..4c3d647c 100644 --- a/queries/admin/account/getAccountById.js +++ b/queries/admin/account/getAccountById.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function getAccountById(user_id) { return runQuery( diff --git a/queries/admin/account/getAccountByUsername.js b/queries/admin/account/getAccountByUsername.js index a777c0ed..30480a71 100644 --- a/queries/admin/account/getAccountByUsername.js +++ b/queries/admin/account/getAccountByUsername.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function getAccountByUsername(username) { return runQuery( diff --git a/queries/admin/account/getAccounts.js b/queries/admin/account/getAccounts.js index e24476df..94f70c35 100644 --- a/queries/admin/account/getAccounts.js +++ b/queries/admin/account/getAccounts.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function getAccounts() { return runQuery( diff --git a/queries/admin/account/updateAccount.js b/queries/admin/account/updateAccount.js index 2137003f..7a337f59 100644 --- a/queries/admin/account/updateAccount.js +++ b/queries/admin/account/updateAccount.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function updateAccount(user_id, data) { return runQuery( diff --git a/queries/admin/website/createWebsite.js b/queries/admin/website/createWebsite.js index 8770d1ce..032a986a 100644 --- a/queries/admin/website/createWebsite.js +++ b/queries/admin/website/createWebsite.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function createWebsite(user_id, data) { return runQuery( diff --git a/queries/admin/website/deleteWebsite.js b/queries/admin/website/deleteWebsite.js index 5d88a3b5..58dc453b 100644 --- a/queries/admin/website/deleteWebsite.js +++ b/queries/admin/website/deleteWebsite.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function deleteWebsite(website_id) { return runQuery( diff --git a/queries/admin/website/getAllWebsites.js b/queries/admin/website/getAllWebsites.js index 03e7d86e..f8d2a3e9 100644 --- a/queries/admin/website/getAllWebsites.js +++ b/queries/admin/website/getAllWebsites.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function getAllWebsites() { let data = await runQuery( diff --git a/queries/admin/website/getUserWebsites.js b/queries/admin/website/getUserWebsites.js index 4db70aab..729ce893 100644 --- a/queries/admin/website/getUserWebsites.js +++ b/queries/admin/website/getUserWebsites.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function getUserWebsites(user_id) { return runQuery( diff --git a/queries/admin/website/getWebsiteById.js b/queries/admin/website/getWebsiteById.js index 231cd899..f4495b93 100644 --- a/queries/admin/website/getWebsiteById.js +++ b/queries/admin/website/getWebsiteById.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function getWebsiteById(website_id) { return runQuery( diff --git a/queries/admin/website/getWebsiteByShareId.js b/queries/admin/website/getWebsiteByShareId.js index 3148b062..5630f89e 100644 --- a/queries/admin/website/getWebsiteByShareId.js +++ b/queries/admin/website/getWebsiteByShareId.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function getWebsiteByShareId(share_id) { return runQuery( diff --git a/queries/admin/website/getWebsiteByUuid.js b/queries/admin/website/getWebsiteByUuid.js index 29e48a06..dd0f24ee 100644 --- a/queries/admin/website/getWebsiteByUuid.js +++ b/queries/admin/website/getWebsiteByUuid.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function getWebsiteByUuid(website_uuid) { return runQuery( diff --git a/queries/admin/website/resetWebsite.js b/queries/admin/website/resetWebsite.js index 8a1c426c..92bb959c 100644 --- a/queries/admin/website/resetWebsite.js +++ b/queries/admin/website/resetWebsite.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function resetWebsite(website_id) { return runQuery(prisma.$queryRaw`delete from session where website_id=${website_id}`); diff --git a/queries/admin/website/updateWebsite.js b/queries/admin/website/updateWebsite.js index 189bd938..9de1a803 100644 --- a/queries/admin/website/updateWebsite.js +++ b/queries/admin/website/updateWebsite.js @@ -1,4 +1,4 @@ -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function updateWebsite(website_id, data) { return runQuery( diff --git a/queries/analytics/event/getEventMetrics.js b/queries/analytics/event/getEventMetrics.js index 41111e71..a3f5c10a 100644 --- a/queries/analytics/event/getEventMetrics.js +++ b/queries/analytics/event/getEventMetrics.js @@ -1,6 +1,6 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; import clickhouse from 'lib/clickhouse'; -import { getDateQuery, getFilterQuery, rawQuery } from 'lib/db/relational'; +import { getDateQuery, getFilterQuery, rawQuery } from 'lib/relational'; import { runAnalyticsQuery } from 'lib/db/db'; export async function getEventMetrics(...args) { diff --git a/queries/analytics/event/getEvents.js b/queries/analytics/event/getEvents.js index 42016fc0..db529e0e 100644 --- a/queries/analytics/event/getEvents.js +++ b/queries/analytics/event/getEvents.js @@ -1,5 +1,5 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; import clickhouse from 'lib/clickhouse'; import { runAnalyticsQuery } from 'lib/db/db'; diff --git a/queries/analytics/event/saveEvent.js b/queries/analytics/event/saveEvent.js index fc73c8b8..52aa8656 100644 --- a/queries/analytics/event/saveEvent.js +++ b/queries/analytics/event/saveEvent.js @@ -1,8 +1,8 @@ import { CLICKHOUSE, KAFKA, RELATIONAL, URL_LENGTH } from 'lib/constants'; import clickhouse from 'lib/clickhouse'; import kafka from 'lib/db/kafka'; -import { prisma, runQuery } from 'lib/db/relational'; -import { runAnalyticsQuery } from 'lib/db/db'; +import { prisma, runQuery } from 'lib/relational'; +import { runAnalyticsQuery } from 'lib/db'; export async function saveEvent(...args) { return runAnalyticsQuery({ diff --git a/queries/analytics/pageview/getPageviewMetrics.js b/queries/analytics/pageview/getPageviewMetrics.js index 3997fd19..92980f3e 100644 --- a/queries/analytics/pageview/getPageviewMetrics.js +++ b/queries/analytics/pageview/getPageviewMetrics.js @@ -1,6 +1,6 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; import clickhouse from 'lib/clickhouse'; -import { parseFilters, rawQuery } from 'lib/db/relational'; +import { parseFilters, rawQuery } from 'lib/relational'; import { runAnalyticsQuery } from 'lib/db/db'; export async function getPageviewMetrics(...args) { diff --git a/queries/analytics/pageview/getPageviewParams.js b/queries/analytics/pageview/getPageviewParams.js index d77d328e..be6e5148 100644 --- a/queries/analytics/pageview/getPageviewParams.js +++ b/queries/analytics/pageview/getPageviewParams.js @@ -1,4 +1,4 @@ -import { parseFilters, rawQuery, runAnalyticsQuery } from 'lib/db/relational'; +import { parseFilters, rawQuery, runAnalyticsQuery } from 'lib/relational'; import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; export async function getPageviewParams(...args) { diff --git a/queries/analytics/pageview/getPageviewStats.js b/queries/analytics/pageview/getPageviewStats.js index 6a786f58..e731318d 100644 --- a/queries/analytics/pageview/getPageviewStats.js +++ b/queries/analytics/pageview/getPageviewStats.js @@ -1,5 +1,5 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { getDateQuery, parseFilters, rawQuery } from 'lib/db/relational'; +import { getDateQuery, parseFilters, rawQuery } from 'lib/relational'; import { runAnalyticsQuery } from 'lib/db/db'; import clickhouse from 'lib/clickhouse'; diff --git a/queries/analytics/pageview/getPageviews.js b/queries/analytics/pageview/getPageviews.js index bccc5bdb..ba9a218c 100644 --- a/queries/analytics/pageview/getPageviews.js +++ b/queries/analytics/pageview/getPageviews.js @@ -1,5 +1,5 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; import { runAnalyticsQuery } from 'lib/db/db'; import clickhouse from 'lib/clickhouse'; diff --git a/queries/analytics/pageview/savePageView.js b/queries/analytics/pageview/savePageView.js index 926b0ad8..56717855 100644 --- a/queries/analytics/pageview/savePageView.js +++ b/queries/analytics/pageview/savePageView.js @@ -2,7 +2,7 @@ import { CLICKHOUSE, KAFKA, RELATIONAL, URL_LENGTH } from 'lib/constants'; import clickhouse from 'lib/clickhouse'; import { runAnalyticsQuery } from 'lib/db/db'; import kafka from 'lib/db/kafka'; -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; export async function savePageView(...args) { return runAnalyticsQuery({ diff --git a/queries/analytics/session/createSession.js b/queries/analytics/session/createSession.js index b6dcd563..a880ce5b 100644 --- a/queries/analytics/session/createSession.js +++ b/queries/analytics/session/createSession.js @@ -1,5 +1,5 @@ import { CLICKHOUSE, KAFKA, RELATIONAL } from 'lib/constants'; -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; import clickhouse from 'lib/clickhouse'; import kafka from 'lib/db/kafka'; import { runAnalyticsQuery } from 'lib/db/db'; diff --git a/queries/analytics/session/getSessionByUuid.js b/queries/analytics/session/getSessionByUuid.js index 6fa6dfba..95302992 100644 --- a/queries/analytics/session/getSessionByUuid.js +++ b/queries/analytics/session/getSessionByUuid.js @@ -1,7 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { prisma, runQuery } from 'lib/db/relational'; +import { prisma, runQuery } from 'lib/relational'; import clickhouse from 'lib/clickhouse'; -import { runAnalyticsQuery } from 'lib/db/db'; +import { runAnalyticsQuery } from 'lib/db'; export async function getSessionByUuid(...args) { return runAnalyticsQuery({ @@ -23,8 +23,9 @@ async function relationalQuery(session_uuid) { async function clickhouseQuery(session_uuid) { const params = [session_uuid]; - return clickhouse.rawQuery( - ` + return clickhouse + .rawQuery( + ` select session_uuid, website_id, @@ -39,6 +40,7 @@ async function clickhouseQuery(session_uuid) { from session where session_uuid = $1 `, - params, - ); + params, + ) + .then(result => clickhouse.findFirst(result)); } diff --git a/queries/analytics/session/getSessionMetrics.js b/queries/analytics/session/getSessionMetrics.js index 6fde5b3d..95a8dc66 100644 --- a/queries/analytics/session/getSessionMetrics.js +++ b/queries/analytics/session/getSessionMetrics.js @@ -1,7 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; import clickhouse from 'lib/clickhouse'; import { runAnalyticsQuery } from 'lib/db/db'; -import { parseFilters, rawQuery } from 'lib/db/relational'; +import { parseFilters, rawQuery } from 'lib/relational'; export async function getSessionMetrics(...args) { return runAnalyticsQuery({ diff --git a/queries/analytics/session/getSessions.js b/queries/analytics/session/getSessions.js index 57806c6f..3fcd1552 100644 --- a/queries/analytics/session/getSessions.js +++ b/queries/analytics/session/getSessions.js @@ -1,7 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; import clickhouse from 'lib/clickhouse'; -import { runAnalyticsQuery } from 'lib/db/db'; -import { prisma, runQuery } from 'lib/db/relational'; +import { runAnalyticsQuery } from 'lib/db'; +import { prisma, runQuery } from 'lib/relational'; export async function getSessions(...args) { return runAnalyticsQuery({ diff --git a/queries/analytics/stats/getActiveVisitors.js b/queries/analytics/stats/getActiveVisitors.js index fa525314..af6d3130 100644 --- a/queries/analytics/stats/getActiveVisitors.js +++ b/queries/analytics/stats/getActiveVisitors.js @@ -1,6 +1,6 @@ import { subMinutes } from 'date-fns'; import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { rawQuery } from 'lib/db/relational'; +import { rawQuery } from 'lib/relational'; import { runAnalyticsQuery } from 'lib/db/db'; import clickhouse from 'lib/clickhouse'; diff --git a/queries/analytics/stats/getWebsiteStats.js b/queries/analytics/stats/getWebsiteStats.js index e4fb4eb8..c8767670 100644 --- a/queries/analytics/stats/getWebsiteStats.js +++ b/queries/analytics/stats/getWebsiteStats.js @@ -1,5 +1,5 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; -import { getDateQuery, getTimestampInterval, parseFilters, rawQuery } from 'lib/db/relational'; +import { getDateQuery, getTimestampInterval, parseFilters, rawQuery } from 'lib/relational'; import { runAnalyticsQuery } from 'lib/db/db'; import clickhouse from 'lib/clickhouse'; From 7e9ed704cc5217255fe5832ac6b52d9699f3c262 Mon Sep 17 00:00:00 2001 From: Brian Cao Date: Thu, 25 Aug 2022 22:23:19 -0700 Subject: [PATCH 5/8] fix ref --- queries/analytics/event/getEventMetrics.js | 2 +- queries/analytics/event/getEvents.js | 2 +- queries/analytics/event/saveEvent.js | 2 +- queries/analytics/pageview/getPageviewMetrics.js | 2 +- queries/analytics/pageview/getPageviewStats.js | 2 +- queries/analytics/pageview/getPageviews.js | 2 +- queries/analytics/pageview/savePageView.js | 4 ++-- queries/analytics/session/createSession.js | 4 ++-- queries/analytics/session/getSessionMetrics.js | 2 +- queries/analytics/stats/getActiveVisitors.js | 2 +- queries/analytics/stats/getWebsiteStats.js | 2 +- 11 files changed, 13 insertions(+), 13 deletions(-) diff --git a/queries/analytics/event/getEventMetrics.js b/queries/analytics/event/getEventMetrics.js index a3f5c10a..531701e8 100644 --- a/queries/analytics/event/getEventMetrics.js +++ b/queries/analytics/event/getEventMetrics.js @@ -1,7 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; import clickhouse from 'lib/clickhouse'; import { getDateQuery, getFilterQuery, rawQuery } from 'lib/relational'; -import { runAnalyticsQuery } from 'lib/db/db'; +import { runAnalyticsQuery } from 'lib/db'; export async function getEventMetrics(...args) { return runAnalyticsQuery({ diff --git a/queries/analytics/event/getEvents.js b/queries/analytics/event/getEvents.js index db529e0e..6bfb6fb8 100644 --- a/queries/analytics/event/getEvents.js +++ b/queries/analytics/event/getEvents.js @@ -1,7 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; import { prisma, runQuery } from 'lib/relational'; import clickhouse from 'lib/clickhouse'; -import { runAnalyticsQuery } from 'lib/db/db'; +import { runAnalyticsQuery } from 'lib/db'; export function getEvents(...args) { return runAnalyticsQuery({ diff --git a/queries/analytics/event/saveEvent.js b/queries/analytics/event/saveEvent.js index 52aa8656..036c8e16 100644 --- a/queries/analytics/event/saveEvent.js +++ b/queries/analytics/event/saveEvent.js @@ -1,6 +1,6 @@ import { CLICKHOUSE, KAFKA, RELATIONAL, URL_LENGTH } from 'lib/constants'; import clickhouse from 'lib/clickhouse'; -import kafka from 'lib/db/kafka'; +import kafka from 'lib/kafka'; import { prisma, runQuery } from 'lib/relational'; import { runAnalyticsQuery } from 'lib/db'; diff --git a/queries/analytics/pageview/getPageviewMetrics.js b/queries/analytics/pageview/getPageviewMetrics.js index 92980f3e..40af97d0 100644 --- a/queries/analytics/pageview/getPageviewMetrics.js +++ b/queries/analytics/pageview/getPageviewMetrics.js @@ -1,7 +1,7 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; import clickhouse from 'lib/clickhouse'; import { parseFilters, rawQuery } from 'lib/relational'; -import { runAnalyticsQuery } from 'lib/db/db'; +import { runAnalyticsQuery } from 'lib/db'; export async function getPageviewMetrics(...args) { return runAnalyticsQuery({ diff --git a/queries/analytics/pageview/getPageviewStats.js b/queries/analytics/pageview/getPageviewStats.js index e731318d..aa1618d3 100644 --- a/queries/analytics/pageview/getPageviewStats.js +++ b/queries/analytics/pageview/getPageviewStats.js @@ -1,6 +1,6 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; import { getDateQuery, parseFilters, rawQuery } from 'lib/relational'; -import { runAnalyticsQuery } from 'lib/db/db'; +import { runAnalyticsQuery } from 'lib/db'; import clickhouse from 'lib/clickhouse'; export async function getPageviewStats(...args) { diff --git a/queries/analytics/pageview/getPageviews.js b/queries/analytics/pageview/getPageviews.js index ba9a218c..fbbd96f8 100644 --- a/queries/analytics/pageview/getPageviews.js +++ b/queries/analytics/pageview/getPageviews.js @@ -1,6 +1,6 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; import { prisma, runQuery } from 'lib/relational'; -import { runAnalyticsQuery } from 'lib/db/db'; +import { runAnalyticsQuery } from 'lib/db'; import clickhouse from 'lib/clickhouse'; export async function getPageviews(...args) { diff --git a/queries/analytics/pageview/savePageView.js b/queries/analytics/pageview/savePageView.js index 56717855..0f93723f 100644 --- a/queries/analytics/pageview/savePageView.js +++ b/queries/analytics/pageview/savePageView.js @@ -1,7 +1,7 @@ import { CLICKHOUSE, KAFKA, RELATIONAL, URL_LENGTH } from 'lib/constants'; import clickhouse from 'lib/clickhouse'; -import { runAnalyticsQuery } from 'lib/db/db'; -import kafka from 'lib/db/kafka'; +import { runAnalyticsQuery } from 'lib/db'; +import kafka from 'lib/kafka'; import { prisma, runQuery } from 'lib/relational'; export async function savePageView(...args) { diff --git a/queries/analytics/session/createSession.js b/queries/analytics/session/createSession.js index a880ce5b..676aeb48 100644 --- a/queries/analytics/session/createSession.js +++ b/queries/analytics/session/createSession.js @@ -1,8 +1,8 @@ import { CLICKHOUSE, KAFKA, RELATIONAL } from 'lib/constants'; import { prisma, runQuery } from 'lib/relational'; import clickhouse from 'lib/clickhouse'; -import kafka from 'lib/db/kafka'; -import { runAnalyticsQuery } from 'lib/db/db'; +import kafka from 'lib/kafka'; +import { runAnalyticsQuery } from 'lib/db'; export async function createSession(...args) { return runAnalyticsQuery({ diff --git a/queries/analytics/session/getSessionMetrics.js b/queries/analytics/session/getSessionMetrics.js index 95a8dc66..e0d5a24c 100644 --- a/queries/analytics/session/getSessionMetrics.js +++ b/queries/analytics/session/getSessionMetrics.js @@ -1,6 +1,6 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; import clickhouse from 'lib/clickhouse'; -import { runAnalyticsQuery } from 'lib/db/db'; +import { runAnalyticsQuery } from 'lib/db'; import { parseFilters, rawQuery } from 'lib/relational'; export async function getSessionMetrics(...args) { diff --git a/queries/analytics/stats/getActiveVisitors.js b/queries/analytics/stats/getActiveVisitors.js index af6d3130..4e61d3eb 100644 --- a/queries/analytics/stats/getActiveVisitors.js +++ b/queries/analytics/stats/getActiveVisitors.js @@ -1,7 +1,7 @@ import { subMinutes } from 'date-fns'; import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; import { rawQuery } from 'lib/relational'; -import { runAnalyticsQuery } from 'lib/db/db'; +import { runAnalyticsQuery } from 'lib/db'; import clickhouse from 'lib/clickhouse'; export async function getActiveVisitors(...args) { diff --git a/queries/analytics/stats/getWebsiteStats.js b/queries/analytics/stats/getWebsiteStats.js index c8767670..081a16ac 100644 --- a/queries/analytics/stats/getWebsiteStats.js +++ b/queries/analytics/stats/getWebsiteStats.js @@ -1,6 +1,6 @@ import { CLICKHOUSE, RELATIONAL } from 'lib/constants'; import { getDateQuery, getTimestampInterval, parseFilters, rawQuery } from 'lib/relational'; -import { runAnalyticsQuery } from 'lib/db/db'; +import { runAnalyticsQuery } from 'lib/db'; import clickhouse from 'lib/clickhouse'; export async function getWebsiteStats(...args) { From f499af5491472c7714cad074a5f0d744f49a0793 Mon Sep 17 00:00:00 2001 From: Brian Cao Date: Thu, 25 Aug 2022 22:43:22 -0700 Subject: [PATCH 6/8] fix refs --- lib/clickhouse.js | 33 +++++++++++++------ lib/kafka.js | 11 +++++-- .../analytics/pageview/getPageviewStats.js | 2 +- 3 files changed, 32 insertions(+), 14 deletions(-) diff --git a/lib/clickhouse.js b/lib/clickhouse.js index ea85d579..7ab06b9e 100644 --- a/lib/clickhouse.js +++ b/lib/clickhouse.js @@ -35,27 +35,27 @@ if (process.env.NODE_ENV !== 'production') { export { clickhouse }; -export function getDateStringQuery(data, unit) { +function getDateStringQuery(data, unit) { return `formatDateTime(${data}, '${CLICKHOUSE_DATE_FORMATS[unit]}')`; } -export function getDateQuery(field, unit, timezone) { +function getDateQuery(field, unit, timezone) { if (timezone) { return `date_trunc('${unit}', ${field}, '${timezone}')`; } return `date_trunc('${unit}', ${field})`; } -export function getDateFormat(date) { +function getDateFormat(date) { return `'${dateFormat(date, 'UTC:yyyy-mm-dd HH:MM:ss')}'`; } -export function getBetweenDates(field, start_at, end_at) { +function getBetweenDates(field, start_at, end_at) { return `${field} between ${getDateFormat(start_at)} and ${getDateFormat(end_at)}`; } -export function getFilterQuery(table, column, filters = {}, params = []) { +function getFilterQuery(table, column, filters = {}, params = []) { const query = Object.keys(filters).reduce((arr, key) => { const filter = filters[key]; @@ -115,7 +115,7 @@ export function getFilterQuery(table, column, filters = {}, params = []) { return query.join('\n'); } -export function parseFilters(table, column, filters = {}, params = [], sessionKey = 'session_id') { +function parseFilters(table, column, filters = {}, params = [], sessionKey = 'session_id') { const { domain, url, event_url, referrer, os, browser, device, country, event_name, query } = filters; @@ -138,7 +138,7 @@ export function parseFilters(table, column, filters = {}, params = [], sessionKe }; } -export function replaceQuery(string, params = []) { +function replaceQuery(string, params = []) { let formattedString = string; params.forEach((a, i) => { @@ -154,7 +154,7 @@ export function replaceQuery(string, params = []) { return formattedString; } -export async function rawQuery(query, params = [], debug = false) { +async function rawQuery(query, params = [], debug = false) { let formattedQuery = replaceQuery(query, params); if (debug || process.env.LOG_QUERY) { @@ -164,7 +164,7 @@ export async function rawQuery(query, params = [], debug = false) { return clickhouse.query(formattedQuery).toPromise(); } -export async function findUnique(data) { +async function findUnique(data) { if (data.length > 1) { throw `${data.length} records found when expecting 1.`; } @@ -172,6 +172,19 @@ export async function findUnique(data) { return data[0] ?? null; } -export async function findFirst(data) { +async function findFirst(data) { return data[0] ?? null; } + +export default { + getDateStringQuery, + getDateQuery, + getDateFormat, + getBetweenDates, + getFilterQuery, + parseFilters, + replaceQuery, + rawQuery, + findUnique, + findFirst, +}; diff --git a/lib/kafka.js b/lib/kafka.js index ab3b3e68..987b9d25 100644 --- a/lib/kafka.js +++ b/lib/kafka.js @@ -44,18 +44,18 @@ let kafkaProducer = null; export { kafka, kafkaProducer }; -export async function getProducer() { +async function getProducer() { const producer = kafka.producer(); await producer.connect(); return producer; } -export function getDateFormat(date) { +function getDateFormat(date) { return dateFormat(date, 'UTC:yyyy-mm-dd HH:MM:ss'); } -export async function sendMessage(params, topic) { +async function sendMessage(params, topic) { await kafkaProducer.send({ topic, messages: [ @@ -67,3 +67,8 @@ export async function sendMessage(params, topic) { acks: 0, }); } + +export default { + getDateFormat, + sendMessage, +}; diff --git a/queries/analytics/pageview/getPageviewStats.js b/queries/analytics/pageview/getPageviewStats.js index aa1618d3..207f109b 100644 --- a/queries/analytics/pageview/getPageviewStats.js +++ b/queries/analytics/pageview/getPageviewStats.js @@ -55,7 +55,7 @@ async function clickhouseQuery( sessionKey = 'session_uuid', ) { const params = [website_id]; - const { pageviewQuery, sessionQuery, joinSession } = clickhouse.parseFilters( + const { pageviewQuery, sessionQuery, joinSession } = parseFilters( 'pageview', null, filters, From 10cc6616c5d4550a8a997a3e3cdc70dca11b8ab1 Mon Sep 17 00:00:00 2001 From: Brian Cao Date: Thu, 25 Aug 2022 23:12:47 -0700 Subject: [PATCH 7/8] add redis placeholder --- lib/session.js | 17 ++++++++++++----- pages/api/collect.js | 35 ----------------------------------- 2 files changed, 12 insertions(+), 40 deletions(-) diff --git a/lib/session.js b/lib/session.js index b90e88f7..8ad83d76 100644 --- a/lib/session.js +++ b/lib/session.js @@ -1,6 +1,8 @@ import { getWebsiteByUuid, getSessionByUuid, createSession } from 'queries'; import { getJsonBody, getClientInfo } from 'lib/request'; import { uuid, isValidUuid, parseToken } from 'lib/crypto'; +import { runAnalyticsQuery } from 'lib/db'; +import { RELATIONAL, CLICKHOUSE } from 'lib/constants'; export async function getSession(req) { const { payload } = getJsonBody(req); @@ -9,7 +11,6 @@ export async function getSession(req) { throw new Error('Invalid request'); } - const { website: website_uuid, hostname, screen, language } = payload; const cache = req.headers['x-umami-cache']; if (cache) { @@ -20,24 +21,30 @@ export async function getSession(req) { } } + const { website: website_uuid, hostname, screen, language } = payload; + if (!isValidUuid(website_uuid)) { throw new Error(`Invalid website: ${website_uuid}`); } - const { userAgent, browser, os, ip, country, device } = await getClientInfo(req, payload); - const website = await getWebsiteByUuid(website_uuid); if (!website) { throw new Error(`Website not found: ${website_uuid}`); } + const { userAgent, browser, os, ip, country, device } = await getClientInfo(req, payload); + const { website_id } = website; const session_uuid = uuid(website_id, hostname, ip, userAgent); - let session = await getSessionByUuid(session_uuid); + // logic placeholder for redis + runAnalyticsQuery({ + [RELATIONAL]: () => {}, + [CLICKHOUSE]: () => {}, + }); - console.log('session here!: ', session); + let session = await getSessionByUuid(session_uuid); if (!session) { try { diff --git a/pages/api/collect.js b/pages/api/collect.js index 4b976319..472af421 100644 --- a/pages/api/collect.js +++ b/pages/api/collect.js @@ -93,38 +93,3 @@ export default async (req, res) => { return send(res, token); }; - -// async function relational(req, res) { -// await useSession(req, res); - -// const { -// session: { website_id, session_id, session_uuid }, -// } = req; - -// const { type, payload } = getJsonBody(req); - -// let { url, referrer, event_name, event_data } = payload; - -// if (process.env.REMOVE_TRAILING_SLASH) { -// url = removeTrailingSlash(url); -// } - -// const event_uuid = uuid(); - -// if (type === 'pageview') { -// await savePageView(website_id, { session_id, session_uuid, url, referrer }); -// } else if (type === 'event') { -// await saveEvent(website_id, { -// event_uuid, -// session_id, -// session_uuid, -// url, -// event_name, -// event_data, -// }); -// } else { -// return badRequest(res); -// } -// } - -// function clickhouse() {} From 818f8721e985412c5001ee8457db7dd68951c3c4 Mon Sep 17 00:00:00 2001 From: Brian Cao Date: Fri, 26 Aug 2022 20:21:53 -0700 Subject: [PATCH 8/8] redis checkpoint --- lib/redis.js | 59 ++++++++++++++++++++++ lib/relational.js | 4 +- package.json | 1 + pages/api/collect.js | 2 +- pages/api/heartbeat.js | 13 ++++- queries/admin/website/createWebsite.js | 9 +++- queries/admin/website/deleteWebsite.js | 39 ++++++++------ queries/analytics/session/createSession.js | 17 +++++-- queries/analytics/session/getSessions.js | 19 ++++--- yarn.lock | 58 ++++++++++++++++++++- 10 files changed, 187 insertions(+), 34 deletions(-) create mode 100644 lib/redis.js diff --git a/lib/redis.js b/lib/redis.js new file mode 100644 index 00000000..97011bed --- /dev/null +++ b/lib/redis.js @@ -0,0 +1,59 @@ +import { createClient } from 'redis'; +import { startOfMonth } from 'date-fns'; +import { getSessions, getAllWebsites } from '/queries'; + +async function getClient() { + const redis = new createClient({ + url: process.env.REDIS_URL, + }); + + await redis.connect(); + + if (process.env.LOG_QUERY) { + redis.on('error', err => console.log('Redis Client Error', err)); + } + + return redis; +} + +let redis = null; + +(async () => { + redis = global.redis || (await getClient()); + + if (process.env.NODE_ENV !== 'production') { + global.redis = redis; + } + + const value = await redis.get('initialized'); + + if (!value) { + await stageData(); + } +})(); + +export default redis; + +async function stageData() { + const sessions = await getSessions([], startOfMonth(new Date()).toUTCString()); + const websites = await getAllWebsites(); + + const sessionUuids = sessions.map(a => { + return { key: `session:${a.session_uuid}`, value: '' }; + }); + const websiteIds = websites.map(a => { + return { key: `website:${a.website_uuid}`, value: Number(a.website_id) }; + }); + + await addRedis(sessionUuids); + await addRedis(websiteIds); + + await redis.set('initialized', 'initialized'); +} + +async function addRedis(ids) { + for (let i = 0; i < ids.length; i++) { + const { key, value } = ids[i]; + await redis.set(key, value); + } +} diff --git a/lib/relational.js b/lib/relational.js index 7d1ed8c2..28de7531 100644 --- a/lib/relational.js +++ b/lib/relational.js @@ -23,7 +23,7 @@ function logQuery(e) { console.log(chalk.yellow(e.params), '->', e.query, chalk.greenBright(`${e.duration}ms`)); } -function getPrismaClient(options) { +function getClient(options) { const prisma = new PrismaClient(options); if (process.env.LOG_QUERY) { @@ -32,7 +32,7 @@ function getPrismaClient(options) { return prisma; } -const prisma = global.prisma || getPrismaClient(options); +const prisma = global.prisma || getClient(options); if (process.env.NODE_ENV !== 'production') { global.prisma = prisma; diff --git a/package.json b/package.json index 6142816a..dfba6613 100644 --- a/package.json +++ b/package.json @@ -97,6 +97,7 @@ "react-tooltip": "^4.2.21", "react-use-measure": "^2.0.4", "react-window": "^1.8.6", + "redis": "^4.3.0", "request-ip": "^3.3.0", "semver": "^7.3.6", "thenby": "^1.3.4", diff --git a/pages/api/collect.js b/pages/api/collect.js index 472af421..c17dd761 100644 --- a/pages/api/collect.js +++ b/pages/api/collect.js @@ -12,7 +12,7 @@ import { uuid } from 'lib/crypto'; export default async (req, res) => { await useCors(req, res); - if (isbot(req.headers['user-agent'])) { + if (isbot(req.headers['user-agent']) && !process.env.DISABLE_BOT_CHECK) { return unauthorized(res); } diff --git a/pages/api/heartbeat.js b/pages/api/heartbeat.js index a53213ad..6218d179 100644 --- a/pages/api/heartbeat.js +++ b/pages/api/heartbeat.js @@ -1,5 +1,16 @@ +import { redis } from 'lib/redis'; import { ok } from 'lib/response'; export default async (req, res) => { - return ok(res, 'nice'); + console.log('------------------------------------------------------------------------------'); + // const redis = new createClient({ + // url: process.env.REDIS_URL, + // }); + + //console.log(redis.get); + + const value = await redis.get('session:77c0b9de-677a-5268-8543-6018f0776a81'); + console.log('complete'); + + return ok(res, value); }; diff --git a/queries/admin/website/createWebsite.js b/queries/admin/website/createWebsite.js index 032a986a..7a589db6 100644 --- a/queries/admin/website/createWebsite.js +++ b/queries/admin/website/createWebsite.js @@ -1,4 +1,5 @@ import { prisma, runQuery } from 'lib/relational'; +import redis from 'lib/redis'; export async function createWebsite(user_id, data) { return runQuery( @@ -12,5 +13,11 @@ export async function createWebsite(user_id, data) { ...data, }, }), - ); + ).then(async res => { + if (process.env.REDIS_URL) { + await redis.set(`website:${res.website_uuid}`, Number(res.website_id)); + } + + return res; + }); } diff --git a/queries/admin/website/deleteWebsite.js b/queries/admin/website/deleteWebsite.js index 58dc453b..13c3160d 100644 --- a/queries/admin/website/deleteWebsite.js +++ b/queries/admin/website/deleteWebsite.js @@ -1,23 +1,30 @@ import { prisma, runQuery } from 'lib/relational'; +import redis from 'lib/redis'; export async function deleteWebsite(website_id) { return runQuery( - prisma.$transaction([ - prisma.pageview.deleteMany({ - where: { session: { website: { website_id } } }, + prisma + .$transaction([ + prisma.pageview.deleteMany({ + where: { session: { website: { website_id } } }, + }), + prisma.event_data.deleteMany({ + where: { event: { session: { website: { website_id } } } }, + }), + prisma.event.deleteMany({ + where: { session: { website: { website_id } } }, + }), + prisma.session.deleteMany({ + where: { website: { website_id } }, + }), + prisma.website.delete({ + where: { website_id }, + }), + ]) + .then(async res => { + if (process.env.REDIS_URL) { + await redis.del(`website:${res.website_uuid}`); + } }), - prisma.event_data.deleteMany({ - where: { event: { session: { website: { website_id } } } }, - }), - prisma.event.deleteMany({ - where: { session: { website: { website_id } } }, - }), - prisma.session.deleteMany({ - where: { website: { website_id } }, - }), - prisma.website.delete({ - where: { website_id }, - }), - ]), ); } diff --git a/queries/analytics/session/createSession.js b/queries/analytics/session/createSession.js index 676aeb48..4cf48566 100644 --- a/queries/analytics/session/createSession.js +++ b/queries/analytics/session/createSession.js @@ -1,8 +1,9 @@ -import { CLICKHOUSE, KAFKA, RELATIONAL } from 'lib/constants'; -import { prisma, runQuery } from 'lib/relational'; import clickhouse from 'lib/clickhouse'; -import kafka from 'lib/kafka'; +import { CLICKHOUSE, KAFKA, RELATIONAL } from 'lib/constants'; import { runAnalyticsQuery } from 'lib/db'; +import kafka from 'lib/kafka'; +import redis from 'lib/redis'; +import { prisma, runQuery } from 'lib/relational'; export async function createSession(...args) { return runAnalyticsQuery({ @@ -23,7 +24,13 @@ async function relationalQuery(website_id, data) { session_id: true, }, }), - ); + ).then(async res => { + if (process.env.REDIS_URL) { + await redis.set(`session:${res.session_uuid}`, ''); + } + + return res; + }); } async function clickhouseQuery( @@ -67,4 +74,6 @@ async function kafkaQuery( }; await kafka.sendKafkaMessage(params, 'session'); + + await redis.set(`session:${session_uuid}`, ''); } diff --git a/queries/analytics/session/getSessions.js b/queries/analytics/session/getSessions.js index 3fcd1552..bcb0f19e 100644 --- a/queries/analytics/session/getSessions.js +++ b/queries/analytics/session/getSessions.js @@ -14,11 +14,15 @@ async function relationalQuery(websites, start_at) { return runQuery( prisma.session.findMany({ where: { - website: { - website_id: { - in: websites, - }, - }, + ...(websites && websites.length > 0 + ? { + website: { + website_id: { + in: websites, + }, + }, + } + : {}), created_at: { gte: start_at, }, @@ -31,7 +35,6 @@ async function clickhouseQuery(websites, start_at) { return clickhouse.rawQuery( ` select - session_id, session_uuid, website_id, created_at, @@ -43,8 +46,8 @@ async function clickhouseQuery(websites, start_at) { "language", country from session - where website_id in (${websites.join[',']} - and created_at >= ${clickhouse.getDateFormat(start_at)}) + where ${websites && websites.length > 0 ? `(website_id in (${websites.join[',']})` : '0 = 0'} + and created_at >= ${clickhouse.getDateFormat(start_at)} `, ); } diff --git a/yarn.lock b/yarn.lock index 98144f5e..4b96ba8e 100644 --- a/yarn.lock +++ b/yarn.lock @@ -1512,6 +1512,40 @@ "@react-spring/shared" "~9.5.0" "@react-spring/types" "~9.5.0" +"@redis/bloom@1.0.2": + version "1.0.2" + resolved "https://registry.yarnpkg.com/@redis/bloom/-/bloom-1.0.2.tgz#42b82ec399a92db05e29fffcdfd9235a5fc15cdf" + integrity sha512-EBw7Ag1hPgFzdznK2PBblc1kdlj5B5Cw3XwI9/oG7tSn85/HKy3X9xHy/8tm/eNXJYHLXHJL/pkwBpFMVVefkw== + +"@redis/client@1.3.0": + version "1.3.0" + resolved "https://registry.yarnpkg.com/@redis/client/-/client-1.3.0.tgz#c62ccd707f16370a2dc2f9e158a28b7da049fa77" + integrity sha512-XCFV60nloXAefDsPnYMjHGtvbtHR8fV5Om8cQ0JYqTNbWcQo/4AryzJ2luRj4blveWazRK/j40gES8M7Cp6cfQ== + dependencies: + cluster-key-slot "1.1.0" + generic-pool "3.8.2" + yallist "4.0.0" + +"@redis/graph@1.0.1": + version "1.0.1" + resolved "https://registry.yarnpkg.com/@redis/graph/-/graph-1.0.1.tgz#eabc58ba99cd70d0c907169c02b55497e4ec8a99" + integrity sha512-oDE4myMCJOCVKYMygEMWuriBgqlS5FqdWerikMoJxzmmTUErnTRRgmIDa2VcgytACZMFqpAOWDzops4DOlnkfQ== + +"@redis/json@1.0.3": + version "1.0.3" + resolved "https://registry.yarnpkg.com/@redis/json/-/json-1.0.3.tgz#a13fde1d22ebff0ae2805cd8e1e70522b08ea866" + integrity sha512-4X0Qv0BzD9Zlb0edkUoau5c1bInWSICqXAGrpwEltkncUwcxJIGEcVryZhLgb0p/3PkKaLIWkjhHRtLe9yiA7Q== + +"@redis/search@1.1.0": + version "1.1.0" + resolved "https://registry.yarnpkg.com/@redis/search/-/search-1.1.0.tgz#7abb18d431f27ceafe6bcb4dd83a3fa67e9ab4df" + integrity sha512-NyFZEVnxIJEybpy+YskjgOJRNsfTYqaPbK/Buv6W2kmFNaRk85JiqjJZA5QkRmWvGbyQYwoO5QfDi2wHskKrQQ== + +"@redis/time-series@1.0.3": + version "1.0.3" + resolved "https://registry.yarnpkg.com/@redis/time-series/-/time-series-1.0.3.tgz#4cfca8e564228c0bddcdf4418cba60c20b224ac4" + integrity sha512-OFp0q4SGrTH0Mruf6oFsHGea58u8vS/iI5+NpYdicaM+7BgqBZH8FFvNZ8rYYLrUO/QRqMq72NpXmxLVNcdmjA== + "@rollup/plugin-buble@^0.21.3": version "0.21.3" resolved "https://registry.npmjs.org/@rollup/plugin-buble/-/plugin-buble-0.21.3.tgz" @@ -2458,6 +2492,11 @@ clone-regexp@^2.1.0: dependencies: is-regexp "^2.0.0" +cluster-key-slot@1.1.0: + version "1.1.0" + resolved "https://registry.yarnpkg.com/cluster-key-slot/-/cluster-key-slot-1.1.0.tgz#30474b2a981fb12172695833052bc0d01336d10d" + integrity sha512-2Nii8p3RwAPiFwsnZvukotvow2rIHM+yQ6ZcBXGHdniadkYGZYiGmkHJIbZPIV9nfv7m/U1IPMVVcAhoWFeklw== + color-convert@^1.9.0, color-convert@^1.9.3: version "1.9.3" resolved "https://registry.npmjs.org/color-convert/-/color-convert-1.9.3.tgz" @@ -3573,6 +3612,11 @@ functions-have-names@^1.2.2: resolved "https://registry.npmjs.org/functions-have-names/-/functions-have-names-1.2.3.tgz" integrity sha512-xckBUXyTIqT97tq2x2AMb+g163b5JFysYk0x4qxNFwbfQkmNZoiRHb6sPzI9/QV33WeuvVYBUIiD4NzNIyqaRQ== +generic-pool@3.8.2: + version "3.8.2" + resolved "https://registry.yarnpkg.com/generic-pool/-/generic-pool-3.8.2.tgz#aab4f280adb522fdfbdc5e5b64d718d3683f04e9" + integrity sha512-nGToKy6p3PAbYQ7p1UlWl6vSPwfwU6TMSWK7TTu+WUY4ZjyZQGniGGt2oNVvyNSpyZYSB43zMXVLcBm08MTMkg== + gensync@^1.0.0-beta.2: version "1.0.0-beta.2" resolved "https://registry.npmjs.org/gensync/-/gensync-1.0.0-beta.2.tgz" @@ -5564,6 +5608,18 @@ redent@^3.0.0: indent-string "^4.0.0" strip-indent "^3.0.0" +redis@^4.3.0: + version "4.3.0" + resolved "https://registry.yarnpkg.com/redis/-/redis-4.3.0.tgz#c01de4cf13821406addb7fcb70afe203266c4c39" + integrity sha512-RXRUor0iU1vizu4viHoUyLpe1ZO/RngZp0V9DyXBHTI+7tC7rEz6Wzn4Sv9v0tTJeqGAzdJ+q5YVbNKKQ5hX9A== + dependencies: + "@redis/bloom" "1.0.2" + "@redis/client" "1.3.0" + "@redis/graph" "1.0.1" + "@redis/json" "1.0.3" + "@redis/search" "1.1.0" + "@redis/time-series" "1.0.3" + redux@^4.0.0, redux@^4.0.4: version "4.2.0" resolved "https://registry.npmjs.org/redux/-/redux-4.2.0.tgz" @@ -6711,7 +6767,7 @@ write-json-file@^4.3.0: sort-keys "^4.0.0" write-file-atomic "^3.0.0" -yallist@^4.0.0: +yallist@4.0.0, yallist@^4.0.0: version "4.0.0" resolved "https://registry.npmjs.org/yallist/-/yallist-4.0.0.tgz" integrity sha512-3wdGidZyq5PB084XLES5TpOSRA3wjXAlIWMhum2kRcv/41Sn2emQ0dycQW4uZXLejwKvg6EsvbdlVL+FYEct7A==