From e9edab7c2932c80461833a42fc6611ecb5bb18da Mon Sep 17 00:00:00 2001 From: amianthus <49116958+SirTenzin@users.noreply.github.com> Date: Fri, 11 Jul 2025 12:22:50 +0100 Subject: [PATCH] =?UTF-8?q?feat:=20=F0=9F=8E=B8=20remove=20supabase=20+=20?= =?UTF-8?q?add=20raw=20events?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../internal/analytics/AnalyticsService.ts | 140 +++++++++--------- .../analytics/internalAnalyticsRouter.ts | 57 +++++-- shared/enums/ErrCode.ts | 3 + 3 files changed, 124 insertions(+), 76 deletions(-) diff --git a/server/src/internal/analytics/AnalyticsService.ts b/server/src/internal/analytics/AnalyticsService.ts index be7344c90..206228cc6 100644 --- a/server/src/internal/analytics/AnalyticsService.ts +++ b/server/src/internal/analytics/AnalyticsService.ts @@ -1,23 +1,25 @@ import { DrizzleCli } from "@/db/initDrizzle.js"; import { ClickHouseClient } from "@clickhouse/client"; -import { events } from "@autumn/shared"; +import { ErrCode, events } from "@autumn/shared"; import { and, eq, sql } from "drizzle-orm"; import { gte, lte } from "drizzle-orm"; import { ExtendedRequest } from "@/utils/models/Request.js"; +import RecaseError from "@/utils/errorUtils.js"; +import { StatusCodes } from "http-status-codes"; export class AnalyticsService { - static clickHouseEnabled = + static clickhouseAvailable = process.env.CLICKHOUSE_URL && process.env.CLICKHOUSE_USERNAME && process.env.CLICKHOUSE_PASSWORD; - static drizzleEnabled = process.env.DATABASE_URL; - static handleEarlyExit = () => { - if (!this.clickHouseEnabled && !this.drizzleEnabled) { - throw new Error( - "Both ClickHouse and Drizzle are disabled, cannot fetch events", - ); + if (!this.clickhouseAvailable) { + throw new RecaseError({ + message: "ClickHouse is disabled, cannot fetch events", + code: ErrCode.ClickHouseDisabled, + statusCode: StatusCodes.SERVICE_UNAVAILABLE, + }); } }; @@ -32,18 +34,20 @@ export class AnalyticsService { return `${year}-${month}-${day} ${hours}:${minutes}:${seconds}`; } - static async getEvents({ + static async getTimeseriesEvents({ req, params, }: { req: ExtendedRequest; params: any; }) { - const { db, clickhouseClient, org } = req; + const { clickhouseClient, org } = req; let startDate = new Date(); const intervalType = params.interval || "day"; + this.handleEarlyExit(); + switch (params.interval) { case "24h": startDate.setHours(startDate.getHours() - 24); @@ -62,18 +66,15 @@ export class AnalyticsService { break; } - this.handleEarlyExit(); - - // Shared count expressions for ClickHouse const countExpressions = params.event_names .map((eventName: string) => - this.clickHouseEnabled + this.clickhouseAvailable ? `countIf(event_name = '${eventName}') AS ${eventName.replace(/[^a-zA-Z0-9]/g, "_")}_count` : `COUNT(*) FILTER (WHERE event_name = '${eventName}') AS ${eventName.replace(/[^a-zA-Z0-9]/g, "_")}_count`, ) .join(",\n "); - if (this.clickHouseEnabled) { + if (this.clickhouseAvailable) { const query = ` SELECT ${ @@ -111,58 +112,63 @@ ORDER BY interval_start ASC;`; const resultJson = await result.json(); return resultJson; - } else { - // Create the interval truncation expression - const getIntervalTrunc = () => - sql`DATE_TRUNC(${ - intervalType === "day" - ? sql`'day'` - : intervalType === "week" - ? sql`'week'` - : intervalType === "month" - ? sql`'month'` - : intervalType === "quarter" - ? sql`'quarter'` - : intervalType === "year" - ? sql`'year'` - : sql`'day'` - }, ${events.timestamp})`; - - // Create Drizzle-specific count expressions - const drizzleCountExpressions = params.event_names.map( - (eventName: string) => - sql`COUNT(*) FILTER (WHERE ${events.event_name} = ${eventName})`.as( - `${eventName.replace(/[^a-zA-Z0-9]/g, "_")}_count`, - ), - ); - - const query = db - .select({ - interval_start: getIntervalTrunc().as("interval_start"), - ...Object.fromEntries( - drizzleCountExpressions.map((expr: any, i: any) => [ - `${params.event_names[i].replace(/[^a-zA-Z0-9]/g, "_")}_count`, - expr, - ]), - ), - }) - .from(events) - .where( - and( - eq(events.org_id, org.id), - eq(events.internal_customer_id, params.customer_id), - gte(events.timestamp, startDate), - lte(events.timestamp, new Date()), - ), - ) - .groupBy(getIntervalTrunc()) - .orderBy(getIntervalTrunc()); - - // console.log("Query:", query); - - const results = await query; - - return { data: results, rows: results.length }; } } + + static async getRawEvents({ + req, + params, + }: { + req: ExtendedRequest; + params: any; + }) { + const { clickhouseClient, org } = req; + + this.handleEarlyExit(); + + let startDate = new Date(); + const intervalType = params.interval || "day"; + + switch (params.interval) { + case "24h": + startDate.setHours(startDate.getHours() - 24); + break; + case "7d": + startDate.setDate(startDate.getDate() - 7); + break; + case "30d": + startDate.setDate(startDate.getDate() - 30); + break; + case "90d": + startDate.setDate(startDate.getDate() - 90); + break; + default: + startDate.setDate(startDate.getDate() - 24); + break; + } + + const query = ` + SELECT timestamp, event_name, value +FROM public_events +WHERE org_id = {organizationId:String} + AND internal_customer_id = {customerId:String} + AND timestamp >= toDateTime({startDate:String}) + AND timestamp < toDateTime({endDate:String}) + + `; + + const result = await (clickhouseClient as ClickHouseClient).query({ + query, + query_params: { + organizationId: org?.id, + customerId: params.customer_id, + startDate: this.formatJsDateToClickHouseDateTime(startDate), + endDate: this.formatJsDateToClickHouseDateTime(new Date()), + }, + }); + + const resultJson = await result.json(); + + return resultJson; + } } diff --git a/server/src/internal/analytics/internalAnalyticsRouter.ts b/server/src/internal/analytics/internalAnalyticsRouter.ts index 4b40232b6..927d01c79 100644 --- a/server/src/internal/analytics/internalAnalyticsRouter.ts +++ b/server/src/internal/analytics/internalAnalyticsRouter.ts @@ -40,13 +40,7 @@ analyticsRouter.post("/events/", async (req: any, res: any) => }); } - console.log("Query input:", { - customer_id, - interval, - event_names, - }); - - const events = await AnalyticsService.getEvents({ + const events = await AnalyticsService.getTimeseriesEvents({ req, params: { customer_id: customer.internal_id, @@ -55,8 +49,6 @@ analyticsRouter.post("/events/", async (req: any, res: any) => }, }); - console.log("Events output:", events); - res.status(200).json({ customer, events, @@ -65,3 +57,50 @@ analyticsRouter.post("/events/", async (req: any, res: any) => }, }), ); + +analyticsRouter.post("/raw/", async (req: any, res: any) => + routeHandler({ + req, + res, + action: "query raw events by customer id", + handler: async () => { + const { db, org, env } = req; + const { interval, customer_id } = req.body; + + if (!customer_id) { + throw new RecaseError({ + message: "Customer ID is required", + code: ErrCode.InvalidRequest, + statusCode: StatusCodes.BAD_REQUEST, + }); + } + + let customer = await CusService.get({ + db, + idOrInternalId: customer_id, + orgId: org.id, + env, + }); + + if (!customer) { + throw new RecaseError({ + message: "Customer not found", + code: ErrCode.CustomerNotFound, + statusCode: StatusCodes.NOT_FOUND, + }); + } + + const events = await AnalyticsService.getRawEvents({ + req, + params: { + customer_id: customer.internal_id, + interval, + }, + }); + + res.status(200).json({ + rawEvents: events, + }); + }, + }), +); diff --git a/shared/enums/ErrCode.ts b/shared/enums/ErrCode.ts index f109fd763..77497ef36 100644 --- a/shared/enums/ErrCode.ts +++ b/shared/enums/ErrCode.ts @@ -162,4 +162,7 @@ export const ErrCode = { SupabaseNotFound: "supabase_not_found", // Entities EntityBalanceNotFound: "entity_balance_not_found", + + // ClickHouse + ClickHouseDisabled: "clickhouse_disabled", };