feat: 🎸 remove supabase + add raw events

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;
+  }
 }

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,
+      });
+    },
+  }),
+);

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",
 };