new consumer

consumer/src/EmailController.ts

@@ -0,0 +1,64 @@
+import { ProjectModel } from "@schema/ProjectSchema";
+import { UserModel } from "@schema/UserSchema";
+import { LimitNotifyModel } from "@schema/broker/LimitNotifySchema";
+import EmailService from '@services/EmailService';
+import { requireEnv } from "@utils/requireEnv";
+import { TProjectLimit } from "@schema/ProjectsLimits";
+
+if (process.env.EMAIL_SERVICE) {
+    EmailService.init(requireEnv('BREVO_API_KEY'));
+}
+
+export async function checkLimitsForEmail(projectCounts: TProjectLimit) {
+
+    const project_id = projectCounts.project_id;
+    const hasNotifyEntry = await LimitNotifyModel.findOne({ project_id });
+    if (!hasNotifyEntry) {
+        await LimitNotifyModel.create({ project_id, limit1: false, limit2: false, limit3: false })
+    }
+
+    if ((projectCounts.visits + projectCounts.events) >= (projectCounts.limit)) {
+
+        if (hasNotifyEntry.limit3 === true) return;
+
+        const project = await ProjectModel.findById(project_id);
+        if (!project) return;
+
+        const owner = await UserModel.findById(project.owner);
+        if (!owner) return;
+
+        if (process.env.EMAIL_SERVICE) await EmailService.sendLimitEmailMax(owner.email, project.name);
+        await LimitNotifyModel.updateOne({ project_id: projectCounts.project_id }, { limit1: true, limit2: true, limit3: true });
+
+    } else if ((projectCounts.visits + projectCounts.events) >= (projectCounts.limit * 0.9)) {
+
+        if (hasNotifyEntry.limit2 === true) return;
+
+        const project = await ProjectModel.findById(project_id);
+        if (!project) return;
+
+        const owner = await UserModel.findById(project.owner);
+        if (!owner) return;
+
+        if (process.env.EMAIL_SERVICE) await EmailService.sendLimitEmail90(owner.email, project.name);
+        await LimitNotifyModel.updateOne({ project_id: projectCounts.project_id }, { limit1: true, limit2: true, limit3: false });
+
+    } else if ((projectCounts.visits + projectCounts.events) >= (projectCounts.limit * 0.5)) {
+
+        if (hasNotifyEntry.limit1 === true) return;
+
+        const project = await ProjectModel.findById(project_id);
+        if (!project) return;
+
+        const owner = await UserModel.findById(project.owner);
+        if (!owner) return;
+
+        if (process.env.EMAIL_SERVICE) await EmailService.sendLimitEmail50(owner.email, project.name);
+        await LimitNotifyModel.updateOne({ project_id: projectCounts.project_id }, { limit1: true, limit2: false, limit3: false });
+
+    }
+
+
+
+
+}

consumer/src/LimitChecker.ts

@@ -0,0 +1,15 @@
+
+
+import { ProjectLimitModel } from '@schema/ProjectsLimits';
+import { MAX_LOG_LIMIT_PERCENT } from '@data/broker/Limits';
+import { checkLimitsForEmail } from './EmailController';
+
+export async function checkLimits(project_id: string) {
+    const projectLimits = await ProjectLimitModel.findOne({ project_id });
+    if (!projectLimits) return false;
+    const TOTAL_COUNT = projectLimits.events + projectLimits.visits;
+    const COUNT_LIMIT = projectLimits.limit;
+    if ((TOTAL_COUNT) > COUNT_LIMIT * MAX_LOG_LIMIT_PERCENT) return false;
+    await checkLimitsForEmail(projectLimits);
+    return true;
+}

consumer/src/index.ts

@@ -0,0 +1,142 @@
+
+import { requireEnv } from '@utils/requireEnv';
+import { connectDatabase } from '@services/DatabaseService';
+import { RedisStreamService } from '@services/RedisStreamService';
+import { ProjectModel } from "@schema/ProjectSchema";
+import { VisitModel } from "@schema/metrics/VisitSchema";
+import { SessionModel } from "@schema/metrics/SessionSchema";
+import { EventModel } from "@schema/metrics/EventSchema";
+import { lookup } from './lookup';
+import { UAParser } from 'ua-parser-js';
+import { checkLimits } from './LimitChecker';
+
+import { ProjectLimitModel } from '@schema/ProjectsLimits';
+import { ProjectCountModel } from '@schema/ProjectsCounts';
+
+connectDatabase(requireEnv('MONGO_CONNECTION_STRING'));
+main();
+
+async function main() {
+
+    await RedisStreamService.connect();
+
+    const stream_name = requireEnv('STREAM_NAME');
+    const group_name = requireEnv('GROUP_NAME') as any; // Checks are inside "startReadingLoop"
+
+    await RedisStreamService.startReadingLoop({
+        stream_name, group_name, consumer_name: `CONSUMER_${process.env.NODE_APP_INSTANCE || 'DEFAULT'}`
+    }, processStreamEntry);
+
+}
+
+async function processStreamEntry(data: Record<string, string>) {
+    try {
+
+        const start = Date.now();
+
+        const eventType = data._type;
+        if (!eventType) return;
+
+        const { pid, sessionHash } = data;
+
+        const project = await ProjectModel.exists({ _id: pid });
+        if (!project) return;
+
+        const canLog = await checkLimits(pid);
+        if (!canLog) return;
+
+        if (eventType === 'event') {
+            await process_event(data, sessionHash);
+        } else if (eventType === 'keep_alive') {
+            await process_keep_alive(data, sessionHash);
+        } else if (eventType === 'visit') {
+            await process_visit(data, sessionHash);
+        }
+
+        const duration = Date.now() - start;
+
+        console.log('Entry processed in', duration, 'ms');
+
+    } catch (ex: any) {
+        console.error('ERROR PROCESSING STREAM EVENT', ex.message);
+    }
+}
+
+async function process_visit(data: Record<string, string>, sessionHash: string) {
+
+    const { pid, ip, website, page, referrer, userAgent, flowHash } = data;
+
+    let referrerParsed;
+    try {
+        referrerParsed = new URL(referrer);
+    } catch (ex) {
+        referrerParsed = { hostname: referrer };
+    }
+
+    const geoLocation = lookup(ip);
+
+    const userAgentParsed = UAParser(userAgent);
+
+    const device = userAgentParsed.device.type;
+
+    await Promise.all([
+        VisitModel.create({
+            project_id: pid, website, page, referrer: referrerParsed.hostname,
+            browser: userAgentParsed.browser.name || 'NO_BROWSER',
+            os: userAgentParsed.os.name || 'NO_OS',
+            device: device ? device : (userAgentParsed.browser.name ? 'desktop' : undefined),
+            session: sessionHash,
+            flowHash,
+            continent: geoLocation[0],
+            country: geoLocation[1],
+        }),
+        ProjectCountModel.updateOne({ project_id: pid }, { $inc: { 'visits': 1 } }, { upsert: true }),
+        ProjectLimitModel.updateOne({ project_id: pid }, { $inc: { 'visits': 1 } })
+    ]);
+
+}
+
+async function process_keep_alive(data: Record<string, string>, sessionHash: string) {
+
+    const { pid, instant, flowHash } = data;
+
+    const existingSession = await SessionModel.findOne({ project_id: pid, session: sessionHash }, { _id: 1 });
+    if (!existingSession) {
+        await ProjectCountModel.updateOne({ project_id: pid }, { $inc: { 'sessions': 1 } }, { upsert: true });
+    }
+
+    if (instant == "true") {
+        await SessionModel.updateOne({ project_id: pid, session: sessionHash, }, {
+            $inc: { duration: 0 },
+            flowHash,
+            updated_at: Date.now()
+        }, { upsert: true });
+    } else {
+        await SessionModel.updateOne({ project_id: pid, session: sessionHash, }, {
+            $inc: { duration: 1 },
+            flowHash,
+            updated_at: Date.now()
+        }, { upsert: true });
+    }
+
+}
+
+async function process_event(data: Record<string, string>, sessionHash: string) {
+
+    const { name, metadata, pid, flowHash } = data;
+
+    let metadataObject;
+    try {
+        if (metadata) metadataObject = JSON.parse(metadata);
+    } catch (ex) {
+        metadataObject = { error: 'Error parsing metadata' }
+    }
+
+    await Promise.all([
+        EventModel.create({ project_id: pid, name, flowHash, metadata: metadataObject, session: sessionHash }),
+        ProjectCountModel.updateOne({ project_id: pid }, { $inc: { 'events': 1 } }, { upsert: true }),
+        ProjectLimitModel.updateOne({ project_id: pid }, { $inc: { 'events': 1 } })
+    ]);
+
+
+}

consumer/src/lookup.ts

@@ -0,0 +1,42 @@
+import fs from 'fs';
+
+const ipsData = JSON.parse(fs.readFileSync('./dist/ipv4-db.json', 'utf8'));
+const countriesData = JSON.parse(fs.readFileSync('./dist/countries-db.json', 'utf8'));
+
+function inRange(ip: string, cidr: string) {
+    const [subnet, mask] = cidr.split('/');
+    const ipBytes = ip.split('.').map(Number);
+    const subnetBytes = subnet.split('.').map(Number);
+
+    const ipInt = (ipBytes[0] << 24) | (ipBytes[1] << 16) | (ipBytes[2] << 8) | ipBytes[3];
+    const subnetInt = (subnetBytes[0] << 24) | (subnetBytes[1] << 16) | (subnetBytes[2] << 8) | subnetBytes[3];
+
+    const maskInt = 0xffffffff << (32 - parseInt(mask));
+
+    return (ipInt & maskInt) === (subnetInt & maskInt);
+}
+
+function getCountryFromId(id: number) {
+    for (const country of countriesData) {
+        if (country[0] == id) {
+            return country;
+        }
+    }
+}
+
+export function lookup(ip: string) {
+    try {
+        const startPiece = parseInt(ip.split('.')[0]);
+        for (const target of ipsData) {
+            const matchingStartPiece = target[0] == startPiece;
+            if (!matchingStartPiece) continue;
+            if (!inRange(ip, target[1])) continue;
+            const country = getCountryFromId(target[2]);
+            return [country[1], country[2]];
+        }
+        return ['??', '??'];
+    } catch (ex) {
+        console.error('ERROR DURING LOOKUP', ex);
+        return ['??', '??'];
+    }
+}