feat: 🎸 tinybird in api

server/src/external/tinybird/initTinybird.ts

@@ -4,6 +4,7 @@ import { createAggregateGroupablePipe } from "./pipes/aggregateGroupablePipe.js"
 import { createAggregatePipe } from "./pipes/aggregatePipe.js";
 import { createAggregateSimplePipe } from "./pipes/aggregateSimplePipe.js";
 import { createListEventNamesPipe } from "./pipes/listEventNamesPipe.js";
+import { createListEventsPaginatedPipe } from "./pipes/listEventsPaginatedPipe.js";
 import { createListEventsPipe } from "./pipes/listEventsPipe.js";
 
 const TINYBIRD_API_URL = process.env.TINYBIRD_API_URL;
@@ -49,6 +50,7 @@ export const tinybirdPipes = tinybirdClient
 			aggregateGroupable: createAggregateGroupablePipe(tinybirdClient),
 			listEvents: createListEventsPipe(tinybirdClient),
 			listEventNames: createListEventNamesPipe(tinybirdClient),
+			listEventsPaginated: createListEventsPaginatedPipe(tinybirdClient),
 		}
 	: null;
 
@@ -94,6 +96,8 @@ export type {
 	AggregateSimplePipeRow,
 	ListEventNamesPipeParams,
 	ListEventNamesPipeRow,
+	ListEventsPaginatedPipeParams,
+	ListEventsPaginatedPipeRow,
 	ListEventsPipeParams,
 	ListEventsPipeRow,
 } from "./pipes/index.js";

server/src/external/tinybird/pipes/index.ts

@@ -19,13 +19,6 @@ export {
 	aggregateSimplePipeResponseSchema,
 	createAggregateSimplePipe,
 } from "./aggregateSimplePipe.js";
-export {
-	createListEventsPipe,
-	type ListEventsPipeParams,
-	type ListEventsPipeRow,
-	listEventsPipeParamsSchema,
-	listEventsPipeResponseSchema,
-} from "./listEventsPipe.js";
 export {
 	createListEventNamesPipe,
 	type ListEventNamesPipeParams,
@@ -33,3 +26,17 @@ export {
 	listEventNamesPipeParamsSchema,
 	listEventNamesPipeResponseSchema,
 } from "./listEventNamesPipe.js";
+export {
+	createListEventsPaginatedPipe,
+	type ListEventsPaginatedPipeParams,
+	type ListEventsPaginatedPipeRow,
+	listEventsPaginatedPipeParamsSchema,
+	listEventsPaginatedPipeResponseSchema,
+} from "./listEventsPaginatedPipe.js";
+export {
+	createListEventsPipe,
+	type ListEventsPipeParams,
+	type ListEventsPipeRow,
+	listEventsPipeParamsSchema,
+	listEventsPipeResponseSchema,
+} from "./listEventsPipe.js";

server/src/external/tinybird/pipes/listEventsPaginatedPipe.ts

@@ -0,0 +1,40 @@
+import type { Tinybird } from "@chronark/zod-bird";
+import { z } from "zod";
+
+/** Response schema for the list_events_paginated pipe */
+export const listEventsPaginatedPipeResponseSchema = z.object({
+	id: z.string(),
+	customer_id: z.string(),
+	event_name: z.string(),
+	timestamp: z.string(),
+	value: z.number().nullable(),
+	properties: z.string().nullable(),
+});
+
+export type ListEventsPaginatedPipeRow = z.infer<
+	typeof listEventsPaginatedPipeResponseSchema
+>;
+
+/** Parameters schema for the list_events_paginated pipe */
+export const listEventsPaginatedPipeParamsSchema = z.object({
+	org_id: z.string(),
+	env: z.string(),
+	start_date: z.string().optional(),
+	end_date: z.string().optional(),
+	customer_id: z.string().optional(),
+	event_names: z.array(z.string()).optional(),
+	limit: z.number().optional(),
+	offset: z.number().optional(),
+});
+
+export type ListEventsPaginatedPipeParams = z.infer<
+	typeof listEventsPaginatedPipeParamsSchema
+>;
+
+/** Creates the list_events_paginated pipe caller */
+export const createListEventsPaginatedPipe = (tb: Tinybird) =>
+	tb.buildPipe({
+		pipe: "list_events_paginated",
+		parameters: listEventsPaginatedPipeParamsSchema,
+		data: listEventsPaginatedPipeResponseSchema,
+	});

server/src/internal/analytics/actions/aggregate.ts

@@ -3,11 +3,14 @@ import {
 	type BillingCycleIntervalEnum,
 	type BillingCycleResult,
 	type ClickHouseResult,
+	ErrCode,
+	RecaseError,
 	type TimeseriesEventsParams,
 } from "@autumn/shared";
 import { UTCDate } from "@date-fns/utc";
 import { addDays, addHours, addMonths, format, sub } from "date-fns";
 import { Decimal } from "decimal.js";
+import { StatusCodes } from "http-status-codes";
 import {
 	type AggregateGroupablePipeRow,
 	type AggregateSimplePipeRow,
@@ -356,6 +359,21 @@ export const aggregate = async ({
 			propertyKey = params.group_by;
 		}
 
+		// Validate property path segments (matches old ClickHouse behavior)
+		if (groupColumn === "property" && propertyKey) {
+			const pathSegments = propertyKey.split(".");
+			for (const segment of pathSegments) {
+				if (!/^[a-zA-Z0-9_]+$/.test(segment)) {
+					throw new RecaseError({
+						message:
+							"Invalid property path. Should only contain alphanumeric and underscore characters.",
+						code: ErrCode.InvalidInputs,
+						statusCode: StatusCodes.BAD_REQUEST,
+					});
+				}
+			}
+		}
+
 		const pipeParams = {
 			org_id: org.id,
 			env,
@@ -375,8 +393,11 @@ export const aggregate = async ({
 
 		const result = await pipes.aggregateGroupable(pipeParams);
 
-		// Extract truncation flag from first row (all rows have the same value)
-		truncated = result.data.length > 0 && result.data[0]._truncated === true;
+		// For external API (enforceGroupLimit), truncated is always false
+		// For internal API, return the actual truncation status from the pipe
+		truncated = params.enforceGroupLimit
+			? false
+			: result.data.length > 0 && result.data[0]._truncated === true;
 
 		formatted = formatGroupableResults({
 			rows: result.data,

server/src/internal/analytics/actions/eventActions.ts

@@ -3,6 +3,7 @@ import { getCountAndSum } from "./getCountAndSum.js";
 import { getEventById } from "./getEventById.js";
 import { getTopEventNames } from "./getTopEventNames.js";
 import { listEventNames } from "./listEventNames.js";
+import { listEventsForApi } from "./listEventsForApi.js";
 import { listRawEvents } from "./listRawEvents.js";
 
 export const eventActions = {
@@ -11,5 +12,6 @@ export const eventActions = {
 	getEventById,
 	getTopEventNames,
 	listEventNames,
+	listEventsForApi,
 	listRawEvents,
 } as const;

server/src/internal/analytics/actions/listEventsForApi.ts

@@ -0,0 +1,104 @@
+import type { ApiEventsListItem } from "@autumn/shared";
+import { getTinybirdPipes } from "@/external/tinybird/initTinybird.js";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+
+/** Converts epoch ms to ClickHouse DateTime string format */
+const epochToDateTime = (epochMs: number): string => {
+	const date = new Date(epochMs);
+	const year = date.getUTCFullYear();
+	const month = String(date.getUTCMonth() + 1).padStart(2, "0");
+	const day = String(date.getUTCDate()).padStart(2, "0");
+	const hours = String(date.getUTCHours()).padStart(2, "0");
+	const minutes = String(date.getUTCMinutes()).padStart(2, "0");
+	const seconds = String(date.getUTCSeconds()).padStart(2, "0");
+	return `${year}-${month}-${day} ${hours}:${minutes}:${seconds}`;
+};
+
+/** Lists events for the external API with offset-based pagination */
+export const listEventsForApi = async ({
+	ctx,
+	params,
+}: {
+	ctx: AutumnContext;
+	params: {
+		customer_id?: string;
+		feature_ids?: string[];
+		custom_range?: { start?: number; end?: number };
+		offset: number;
+		limit: number;
+	};
+}) => {
+	const pipes = getTinybirdPipes();
+	const { org, env } = ctx;
+
+	// Convert epoch ms to DateTime strings (if provided)
+	const startDate = params.custom_range?.start
+		? epochToDateTime(params.custom_range.start)
+		: undefined;
+	const endDate = params.custom_range?.end
+		? epochToDateTime(params.custom_range.end)
+		: undefined;
+
+	// Fetch N+1 for has_more calculation
+	const fetchLimit = params.limit + 1;
+
+	ctx.logger.debug("Listing events for API via Tinybird", {
+		customerId: params.customer_id,
+		featureIds: params.feature_ids,
+		startDate,
+		endDate,
+		offset: params.offset,
+		limit: params.limit,
+	});
+
+	const startTime = performance.now();
+	const result = await pipes.listEventsPaginated({
+		org_id: org.id,
+		env,
+		start_date: startDate,
+		end_date: endDate,
+		customer_id: params.customer_id,
+		event_names: params.feature_ids,
+		limit: fetchLimit,
+		offset: params.offset,
+	});
+
+	const queryDuration = performance.now() - startTime;
+	const hasMore = result.data.length > params.limit;
+	const rows = hasMore ? result.data.slice(0, params.limit) : result.data;
+
+	// Transform to API format
+	const list: ApiEventsListItem[] = rows.map((row) => {
+		let properties = {};
+		if (row.properties) {
+			try {
+				properties = JSON.parse(row.properties);
+			} catch {
+				// Invalid JSON, use empty object
+			}
+		}
+
+		return {
+			id: row.id,
+			timestamp: new Date(row.timestamp).getTime(),
+			feature_id: row.event_name,
+			customer_id: row.customer_id,
+			value: row.value ?? 0,
+			properties,
+		};
+	});
+
+	ctx.logger.debug("Events list result", {
+		queryMs: Math.round(queryDuration),
+		rowCount: list.length,
+		hasMore,
+	});
+
+	return {
+		list,
+		has_more: hasMore,
+		total: list.length,
+		offset: params.offset,
+		limit: params.limit,
+	};
+};

server/src/internal/analytics/internalAnalyticsRouter.ts

@@ -1,13 +1,13 @@
 import { Hono } from "hono";
 import type { HonoEnv } from "@/honoUtils/HonoEnv.js";
 import { handleGetEventNames } from "./internalHandlers/handleGetEventNames.js";
+import { handleInternalAggregateEvents } from "./internalHandlers/handleInternalAggregateEvents.js";
+import { handleInternalListRawEvents } from "./internalHandlers/handleInternalListRawEvents.js";
 import { handleListEventNames } from "./internalHandlers/handleListEventNames.js";
-import { handleQueryEvents } from "./internalHandlers/handleQueryEvents.js";
-import { handleQueryRawEvents } from "./internalHandlers/handleQueryRawEvents.js";
 
 export const internalAnalyticsRouter = new Hono<HonoEnv>();
 
 internalAnalyticsRouter.get("/event_names", ...handleGetEventNames);
 internalAnalyticsRouter.get("/event_names/list", ...handleListEventNames);
-internalAnalyticsRouter.post("/events", ...handleQueryEvents);
-internalAnalyticsRouter.post("/raw", ...handleQueryRawEvents);
+internalAnalyticsRouter.post("/events", ...handleInternalAggregateEvents);
+internalAnalyticsRouter.post("/raw", ...handleInternalListRawEvents);

server/src/internal/analytics/internalHandlers/handleGetEventNames.ts

@@ -1,7 +1,7 @@
 import { type Feature, FeatureType } from "@autumn/shared";
 import { createRoute } from "@/honoMiddlewares/routeHandler.js";
-import { eventActions } from "../actions/index.js";
 import { AnalyticsService } from "../AnalyticsService.js";
+import { eventActions } from "../actions/eventActions.js";
 
 /**
  * Get top event names for the organization
@@ -9,7 +9,7 @@ import { AnalyticsService } from "../AnalyticsService.js";
 export const handleGetEventNames = createRoute({
 	handler: async (c) => {
 		const ctx = c.get("ctx");
-		const { org, env, features } = ctx;
+		const { features } = ctx;
 
 		AnalyticsService.handleEarlyExit();
 

server/src/internal/analytics/internalHandlers/handleInternalAggregateEvents.ts

@@ -10,9 +10,9 @@ import { z } from "zod/v4";
 import { createRoute } from "@/honoMiddlewares/routeHandler.js";
 import { CusService } from "@/internal/customers/CusService.js";
 import { AnalyticsService } from "../AnalyticsService.js";
-import { eventActions } from "../actions/index.js";
+import { eventActions } from "../actions/eventActions.js";
 
-const QueryEventsSchema = z.object({
+const InternalAggregateEventsSchema = z.object({
 	interval: z.string().nullish(),
 	event_names: z.array(z.string()),
 	customer_id: z.string().optional(),
@@ -24,8 +24,8 @@ const QueryEventsSchema = z.object({
 /**
  * Query events by customer ID
  */
-export const handleQueryEvents = createRoute({
-	body: QueryEventsSchema,
+export const handleInternalAggregateEvents = createRoute({
+	body: InternalAggregateEventsSchema,
 	handler: async (c) => {
 		const ctx = c.get("ctx");
 		const { db, org, env, features } = ctx;

server/src/internal/analytics/internalHandlers/handleInternalListRawEvents.ts

@@ -3,10 +3,9 @@ import { StatusCodes } from "http-status-codes";
 import { z } from "zod/v4";
 import { createRoute } from "@/honoMiddlewares/routeHandler.js";
 import { CusService } from "@/internal/customers/CusService.js";
-import { AnalyticsService } from "../AnalyticsService.js";
-import { eventActions } from "../actions/index.js";
+import { eventActions } from "../actions/eventActions.js";
 
-const QueryRawEventsSchema = z.object({
+const InternalListRawEventsSchema = z.object({
 	interval: z.string().nullish(),
 	customer_id: z.string().nullish(),
 });
@@ -14,15 +13,13 @@ const QueryRawEventsSchema = z.object({
 /**
  * Query raw events by customer ID
  */
-export const handleQueryRawEvents = createRoute({
-	body: QueryRawEventsSchema,
+export const handleInternalListRawEvents = createRoute({
+	body: InternalListRawEventsSchema,
 	handler: async (c) => {
 		const ctx = c.get("ctx");
 		const { db, org, env } = ctx;
 		const { interval, customer_id } = c.req.valid("json");
 
-		AnalyticsService.handleEarlyExit();
-
 		let aggregateAll = false;
 		let customer: FullCustomer | undefined;
 

server/src/internal/analytics/internalHandlers/handleListEventNames.ts

@@ -1,7 +1,7 @@
 import { z } from "zod/v4";
 import { createRoute } from "@/honoMiddlewares/routeHandler.js";
 import { AnalyticsService } from "../AnalyticsService.js";
-import { eventActions } from "../actions/index.js";
+import { eventActions } from "../actions/eventActions.js";
 
 const ListEventNamesSchema = z.object({
 	limit: z.coerce.number().optional(),

server/src/internal/analytics/legacyAnalyticsRouter.ts

@@ -1,7 +1,7 @@
 import { Hono } from "hono";
 import type { HonoEnv } from "../../honoUtils/HonoEnv.js";
-import { handleAggregateEvents } from "../events/handlers/handleAggregateEvents.js";
+import { handleExternalAggregateEvents } from "../events/handlers/handleExternalAggregateEvents.js";
 
 export const legacyAnalyticsRouter = new Hono<HonoEnv>();
 
-legacyAnalyticsRouter.post("", ...handleAggregateEvents);
+legacyAnalyticsRouter.post("", ...handleExternalAggregateEvents);

server/src/internal/events/eventsRouter.ts

@@ -1,9 +1,9 @@
 import { Hono } from "hono";
 import type { HonoEnv } from "../../honoUtils/HonoEnv.js";
-import { handleAggregateEvents } from "./handlers/handleAggregateEvents.js";
-import { handleListEvents } from "./handlers/handleListEvents.js";
+import { handleExternalAggregateEvents } from "./handlers/handleExternalAggregateEvents.js";
+import { handleExternalListEvents } from "./handlers/handleExternalListEvents.js";
 
 export const eventsRouter = new Hono<HonoEnv>();
 
-eventsRouter.post("aggregate", ...handleAggregateEvents);
-eventsRouter.post("list", ...handleListEvents);
+eventsRouter.post("aggregate", ...handleExternalAggregateEvents);
+eventsRouter.post("list", ...handleExternalListEvents);

server/src/internal/events/handlers/handleExternalAggregateEvents.ts

@@ -6,9 +6,9 @@ import {
 	RecaseError,
 } from "@autumn/shared";
 import { StatusCodes } from "http-status-codes";
+import { eventActions } from "@/internal/analytics/actions/eventActions.js";
 import { CusService } from "@/internal/customers/CusService";
 import { createRoute } from "../../../honoMiddlewares/routeHandler";
-import { EventsAggregationService } from "../EventsAggregationService";
 import {
 	backfillMissingGroupValues,
 	buildGroupedTimeseries,
@@ -16,7 +16,7 @@ import {
 	convertPeriodsToEpoch,
 } from "../eventUtils.js";
 
-export const handleAggregateEvents = createRoute({
+export const handleExternalAggregateEvents = createRoute({
 	body: EventsAggregateParamsSchema,
 	handler: async (c) => {
 		const ctx = c.get("ctx");
@@ -47,8 +47,8 @@ export const handleAggregateEvents = createRoute({
 
 		const featureIds = Array.isArray(feature_id) ? feature_id : [feature_id];
 
-		const [events, total] = await Promise.all([
-			EventsAggregationService.getTimeseriesEvents({
+		const [eventsResult, total] = await Promise.all([
+			eventActions.aggregate({
 				ctx,
 				params: {
 					aggregateAll: false,
@@ -60,9 +60,10 @@ export const handleAggregateEvents = createRoute({
 					group_by,
 					bin_size: bin_size ?? "day",
 					custom_range,
+					enforceGroupLimit: true,
 				},
 			}),
-			EventsAggregationService.getTotalEvents({
+			eventActions.getCountAndSum({
 				ctx,
 				params: {
 					aggregateAll: false,
@@ -76,6 +77,8 @@ export const handleAggregateEvents = createRoute({
 			}),
 		]);
 
+		const events = eventsResult.formatted;
+
 		if (!events) {
 			throw new RecaseError({
 				message: "No events found",

server/src/internal/events/handlers/handleExternalListEvents.ts

@@ -0,0 +1,33 @@
+import type { ApiEventsListResponse } from "@autumn/shared";
+import { ApiEventsListParamsSchema } from "@autumn/shared";
+import { createRoute } from "@/honoMiddlewares/routeHandler";
+import { eventActions } from "@/internal/analytics/actions/eventActions.js";
+
+export const handleExternalListEvents = createRoute({
+	body: ApiEventsListParamsSchema,
+	handler: async (c) => {
+		const ctx = c.get("ctx");
+		const validatedParams = ApiEventsListParamsSchema.parse(
+			c.req.valid("json"),
+		);
+
+		const featureIds = validatedParams.feature_id
+			? Array.isArray(validatedParams.feature_id)
+				? validatedParams.feature_id
+				: [validatedParams.feature_id]
+			: undefined;
+
+		const result = await eventActions.listEventsForApi({
+			ctx,
+			params: {
+				customer_id: validatedParams.customer_id,
+				feature_ids: featureIds,
+				custom_range: validatedParams.custom_range,
+				offset: validatedParams.offset,
+				limit: validatedParams.limit,
+			},
+		});
+
+		return c.json<ApiEventsListResponse>(result);
+	},
+});

server/src/internal/events/handlers/handleListEvents.ts

@@ -1,19 +0,0 @@
-import type { ApiEventsListResponse } from "@autumn/shared";
-import { ApiEventsListParamsSchema } from "@autumn/shared";
-import { createRoute } from "@/honoMiddlewares/routeHandler";
-import { EventListService } from "../EventListService";
-
-export const handleListEvents = createRoute({
-	body: ApiEventsListParamsSchema,
-	handler: async (c) => {
-		const ctx = c.get("ctx");
-		const bodyParams = c.req.valid("json");
-
-		const result = await EventListService.getEvents({
-			ctx,
-			params: bodyParams,
-		});
-
-		return c.json<ApiEventsListResponse>(result);
-	},
-});

server/tests/integration/balances/track/track-tinybird-migration.test.ts

@@ -3,7 +3,7 @@ import { TestFeature } from "@tests/setup/v2Features";
 import { items } from "@tests/utils/fixtures/items";
 import { products } from "@tests/utils/fixtures/products";
 import { initScenario, s } from "@tests/utils/testInitUtils/initScenario";
-import { eventActions } from "@/internal/analytics/actions/index.js";
+import { eventActions } from "@/internal/analytics/actions/eventActions.js";
 import { generateId, timeout } from "@/utils/genUtils.js";
 
 const free = products.base({

server/tinybird/pipes/list_events_paginated.pipe

@@ -0,0 +1,36 @@
+DESCRIPTION >
+    Lists raw events with offset-based pagination for external API.
+    Supports filtering by customer_id, event_names (array), and optional date range.
+
+TOKEN "list_events_paginated_read" READ
+
+NODE endpoint
+TYPE endpoint
+SQL >
+    %
+    SELECT
+        id,
+        customer_id,
+        event_name,
+        timestamp,
+        value,
+        properties
+    FROM events_by_timestamp_mv
+    WHERE
+        org_id = {{ String(org_id, '') }}
+        AND env = {{ String(env, 'test') }}
+        {% if defined(start_date) and String(start_date, '') != '' %}
+        AND timestamp >= toDateTime64({{ String(start_date) }}, 6)
+        {% end %}
+        {% if defined(end_date) and String(end_date, '') != '' %}
+        AND timestamp <= toDateTime64({{ String(end_date) }}, 6)
+        {% end %}
+        {% if defined(customer_id) and String(customer_id, '') != '' %}
+        AND customer_id = {{ String(customer_id) }}
+        {% end %}
+        {% if defined(event_names) %}
+        AND event_name IN {{ Array(event_names, 'String') }}
+        {% end %}
+    ORDER BY timestamp DESC, id DESC
+    LIMIT {{ Int32(limit, 101) }}
+    OFFSET {{ Int32(offset, 0) }}

shared/models/eventModels/eventTypes.ts

@@ -29,6 +29,7 @@ export type TimeseriesEventsParams = TotalEventsParams & {
 	group_by?: string;
 	no_count?: boolean;
 	timezone?: string;
+	enforceGroupLimit?: boolean;
 };
 
 export type CalculateDateRangeParams = Omit<