feat: push_event and query_instant

Author: abcxff

modules/analytics/db/migrations/20240819042229_init/migration.sql

@@ -0,0 +1,13 @@
+-- CreateTable
+CREATE TABLE "Event" (
+    "id" UUID NOT NULL,
+    "timestamp" TIMESTAMPTZ NOT NULL,
+    "name" TEXT NOT NULL,
+    "metadata" JSONB
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "Event_id_key" ON "Event"("id");
+
+-- CreateIndex
+CREATE INDEX "Event_name_time_idx" ON "Event"("name", "timestamp" DESC);

modules/analytics/db/schema.prisma

@@ -12,6 +12,5 @@ model Event {
 
     // in the init migration, we add the timescale extension and call create_hypertable().
 
-    @@index(fields: [name, timestamp(sort: Desc)], map: "event_name_time_idx")
-    @@map("event")
+    @@index(fields: [name, timestamp(sort: Desc)], map: "Event_name_time_idx")
 }

modules/analytics/module.json

@@ -11,6 +11,9 @@
 	"scripts": {
 		"push_event": {
 			"public": true
+		},
+		"query_instant": {
+			"public": true
 		}
 	},
 	"errors": {},

modules/analytics/scripts/push_event.ts

@@ -1,5 +1,5 @@
 import { ScriptContext } from "../module.gen.ts";
-import { createHypertable } from "../utils/hypertable_init.ts";
+import { checkHypertable } from "../utils/hypertable_init.ts";
 
 export interface Request {
     name: string,
@@ -16,7 +16,7 @@ export async function run(
     ctx: ScriptContext,
     req: Request,
 ): Promise<Response> {
-	createHypertable(ctx);
+    checkHypertable(ctx);
     const timestamp = req.timestampOverride ? new Date(req.timestampOverride) : new Date();
     const event = await ctx.db.event.create({
         data: {

modules/analytics/scripts/query_instant.ts

@@ -0,0 +1,48 @@
+import { ScriptContext } from "../module.gen.ts";
+import { checkHypertable } from "../utils/hypertable_init.ts";
+import { stringifyFilters } from "../utils/stringify_filters.ts";
+import { AggregationMethod, Filter } from "../utils/types.ts";
+
+export interface Request {
+    event: string;
+    aggregate: AggregationMethod;
+    filters: Filter[]
+    groupBy: string[];
+    startAt: number;
+    stopAt: number;
+}
+
+export interface Response {
+    results: { groups: Record<string, any>, count: number}[]
+}
+
+export async function run(
+    ctx: ScriptContext,
+    req: Request,
+): Promise<Response> {
+	checkHypertable(ctx);
+
+    const props = req.groupBy.map((col) => `metadata->>'${col}'`);
+
+    // A query that counts the amount of events in the database, per name (should return an array of counts per name)
+    // the name isn't an actual field but instead a value in the metadata field
+    const result = await ctx.db.$queryRawUnsafe(`
+        SELECT ${req.groupBy.map(col => `metadata->>'${col}' as _${col}`).join(', ')}, COUNT(*) as count
+        FROM "${ctx.dbSchema}"."Event"
+        WHERE name = '${req.event}'
+            AND timestamp >= '${new Date(req.startAt).toISOString()}'
+            AND timestamp <= '${new Date(req.stopAt).toISOString()}'
+            ${req.filters.length ? " AND " + stringifyFilters(req.filters) : ""}
+        GROUP BY ${props.join(', ')}
+        ORDER BY ${props.join(', ')}
+    `) as any;
+
+    return { 
+        results: result.map((e: any) => ({
+            // TODO: optimize
+            groups: props.reduce<Record<string, any>>((acc, k) => (acc[k] = e["_" + k], acc), {}),
+            count: e.count
+    }))
+    }
+}
+

modules/analytics/utils/hypertable_init.ts

@@ -1,10 +1,10 @@
 import { ScriptContext } from "../module.gen.ts";
 
 let hasDefinitelyRun = false;
-export const createHypertable = async (ctx: ScriptContext) => {
+export const checkHypertable = async (ctx: ScriptContext) => {
     if (hasDefinitelyRun) return;
 
-    await ctx.db.$queryRaw`SELECT create_hypertable('event', 'timestamp');`;
+    // await ctx.db.$queryRaw`SELECT create_hypertable('event', 'timestamp');`;
 
     hasDefinitelyRun = true;
 }

modules/analytics/utils/stringify_filters.ts

@@ -0,0 +1,12 @@
+import { Filter } from "./types.ts";
+
+export const stringifyFilters = (filters: Filter[]) => filters.map((filter: Filter) => {
+    if ("greaterThan" in filter) return "(metadata->>'" + filter.greaterThan.key + "')::int" + " > " + filter.greaterThan.value;
+    if ("lessThan" in filter) return "(metadata->>'" + filter.lessThan.key + "')::int" + " < " + filter.lessThan.value;
+    if ("equals" in filter) return "(metadata->>'" + filter.equals.key + "')::int" + " = " + filter.equals.value;
+    if ("notEquals" in filter) return "(metadata->>'" + filter.notEquals.key + "')::int" + " != " + filter.notEquals.value;
+    if ("greaterThanOrEquals" in filter) return "(metadata->>'" + filter.greaterThanOrEquals.key + "')::int" + " >= " + filter.greaterThanOrEquals.value;
+    if ("lessThanOrEquals" in filter) return "(metadata->>'" + filter.lessThanOrEquals.key + "')::int" + " <= " + filter.lessThanOrEquals.value;
+
+    throw new Error("Unknown filter type");
+}).join(' AND ');

modules/analytics/utils/types.ts

@@ -0,0 +1,10 @@
+export type AggregationMethod = { count: {} } |
+    { averageByKey: string } |
+    { sumByKey: string };
+
+export type Filter = { greaterThan: { key: string, value: number } } |
+    { lessThan: { key: string, value: number } } |
+    { equals: { key: string, value: number } } |
+    { notEquals: { key: string, value: number } } |
+    { greaterThanOrEquals: { key: string, value: number } } |
+    { lessThanOrEquals: { key: string, value: number } };