feat(analytics): implement push_event and hypertable init

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

@@ -9,6 +9,12 @@
 		"ABCxFF"
 	],
 	"scripts": {
+		"push_event": {
+			"public": true
+		},
+		"query_instant": {
+			"public": true
+		}
 	},
 	"errors": {},
 	"dependencies": {}

modules/analytics/scripts/push_event.ts

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

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

@@ -0,0 +1,10 @@
+import { ScriptContext } from "../module.gen.ts";
+
+let hasDefinitelyRun = false;
+export const checkHypertable = async (ctx: ScriptContext) => {
+    if (hasDefinitelyRun) return;
+
+    // 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 } };