feat: 🎸 plan_id in tb schemas

server/src/internal/balances/track/v3/runPostgresTrackV3.ts

@@ -6,6 +6,7 @@ import {
 	buildEventInfo,
 	initEvent,
 } from "@/internal/balances/events/initEvent.js";
+import { resolveInternalProductIdForEvent } from "../../events/resolveInternalProductIdForEvent.js";
 import {
 	deductionToTrackResponseV2,
 	executePostgresDeductionV2,
@@ -53,6 +54,11 @@ export const runPostgresTrackV3 = async ({
 		mutationLogs,
 	});
 
+	const internalProductId = resolveInternalProductIdForEvent({
+		fullSubject: updatedFullSubject,
+		mutationLogs,
+	});
+
 	if (!body.skip_event && !body.idempotency_key) {
 		const eventInfo = buildEventInfo(body);
 		const event = initEvent({
@@ -62,6 +68,7 @@ export const runPostgresTrackV3 = async ({
 			internalEntityId: updatedFullSubject.internalEntityId,
 			customerId: body.customer_id,
 			entityId: body.entity_id,
+			internalProductId,
 			deductions,
 		});
 

server/src/internal/balances/track/v3/runRedisTrackV3.ts

@@ -7,6 +7,7 @@ import type {
 import { tryCatch } from "@autumn/shared";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { globalEventBatchingManager } from "@/internal/balances/events/EventBatchingManager.js";
+import { resolveInternalProductIdForEvent } from "@/internal/balances/events/resolveInternalProductIdForEvent.js";
 import {
 	buildEventInfo,
 	initEvent,
@@ -56,11 +57,13 @@ const queueEvent = ({
 	body,
 	fullSubject,
 	deductions,
+	internalProductId,
 }: {
 	ctx: AutumnContext;
 	body: TrackParams;
 	fullSubject: FullSubject;
 	deductions: TrackDeduction[];
+	internalProductId: string | null;
 }): void => {
 	if (body.skip_event) return;
 
@@ -75,6 +78,7 @@ const queueEvent = ({
 			customerId: body.customer_id,
 			entityId: body.entity_id,
 			deductions,
+			internalProductId,
 		}),
 	);
 };
@@ -139,7 +143,12 @@ export const runRedisTrackV3 = async ({
 		mutationLogs,
 	});
 
-	queueEvent({ ctx, body, fullSubject, deductions });
+	const internalProductId = resolveInternalProductIdForEvent({
+		fullSubject: updatedFullSubject,
+		mutationLogs,
+	});
+
+	queueEvent({ ctx, body, fullSubject, deductions, internalProductId });
 
 	const { balance, balances } = await deductionToTrackResponseV2({
 		ctx,

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

@@ -24,7 +24,6 @@ export const handleExternalAggregateEvents = createRoute({
 	body: EventsAggregateParamsSchema,
 	handler: async (c) => {
 		const ctx = c.get("ctx");
-		const { db, org, env } = ctx;
 		const {
 			customer_id,
 			entity_id,
@@ -56,12 +55,13 @@ export const handleExternalAggregateEvents = createRoute({
 
 		const featureIds = Array.isArray(feature_id) ? feature_id : [feature_id];
 
-		// Map special $-prefixed group_by operators to their column names
 		let resolvedGroupBy = group_by;
 		if (group_by === "$customer_id") {
 			resolvedGroupBy = "customer_id";
 		} else if (group_by === "$entity_id") {
 			resolvedGroupBy = "entity_id";
+		} else if (group_by === "$plan_id") {
+			resolvedGroupBy = "plan_id";
 		}
 
 		const [eventsResult, total] = await Promise.all([

server/tinybird/connections/firecrawl_backfill_apikeyid.connection

@@ -1,4 +0,0 @@
-TYPE s3
-S3_ARN {{ tb_secret("s3_role_arn_firecrawl_backfill_apikeyid_75f684b3") }}
-S3_REGION us-west-2
-# Learn more at https://www.tinybird.co/docs/forward/get-data-in/connectors/s3#s3-connection-settings

server/tinybird/datasources/events.datasource

@@ -15,6 +15,7 @@ SCHEMA >
     `set_usage` Nullable(UInt8) `json:$.set_usage`,
     `entity_id` Nullable(String) `json:$.entity_id`,
     `internal_entity_id` Nullable(String) `json:$.internal_entity_id`,
+    `internal_product_id` Nullable(String) `json:$.internal_product_id`,
     `customer_id` String `json:$.customer_id`,
     `properties` JSON `json:$.properties`,
     `deductions` JSON `json:$.deductions`

server/tinybird/materializations/events_by_timestamp_mv.datasource

@@ -14,6 +14,7 @@ SCHEMA >
     `properties` Nullable(String),
     `idempotency_key` Nullable(String),
     `entity_id` String DEFAULT '',
+    `internal_product_id` Nullable(String),
     `deductions` Nullable(String)
 
 ENGINE "MergeTree"

server/tinybird/materializations/events_by_timestamp_mv_pipe.pipe

@@ -15,6 +15,7 @@ SQL >
         properties,
         idempotency_key,
         coalesce(entity_id, '') as entity_id,
+        internal_product_id,
         toString(deductions) as deductions
     FROM events
     ORDER BY timestamp DESC, id DESC

server/tinybird/materializations/events_hourly_mv.datasource

@@ -11,6 +11,7 @@ SCHEMA >
     `customer_id` String,
     `event_name` String,
     `entity_id` String DEFAULT '',
+    `internal_product_id` Nullable(String),
     `hour` DateTime,
     `properties` JSON,
     `total_value` Float64,

server/tinybird/materializations/events_hourly_mv_pipe.pipe

@@ -10,12 +10,13 @@ SQL >
         customer_id,
         event_name,
         coalesce(entity_id, '') as entity_id,
+        internal_product_id,
         toStartOfHour(timestamp) as hour,
         CAST(coalesce(properties, '{}') AS JSON) as properties,
         sum(toFloat64(coalesce(value, 1))) as total_value,
         count() as event_count
     FROM events
-    GROUP BY org_id, env, customer_id, event_name, entity_id, hour, properties
+    GROUP BY org_id, env, customer_id, event_name, entity_id, internal_product_id, hour, properties
 
 TYPE materialized
 DATASOURCE events_hourly_mv

server/tinybird/materializations/events_hourly_no_properties_two_mv.datasource

@@ -11,6 +11,7 @@ SCHEMA >
     `customer_id` String,
     `event_name` String,
     `entity_id` String DEFAULT '',
+    `internal_product_id` Nullable(String),
     `hour` DateTime,
     `total_value` Float64,
     `event_count` UInt64

server/tinybird/materializations/events_hourly_no_properties_two_mv_pipe.pipe

@@ -11,11 +11,12 @@ SQL >
         customer_id,
         event_name,
         coalesce(entity_id, '') as entity_id,
+        internal_product_id,
         toStartOfHour(timestamp) as hour,
         sum(toFloat64(coalesce(value, 1))) as total_value,
         count() as event_count
     FROM events
-    GROUP BY org_id, env, customer_id, event_name, entity_id, hour
+    GROUP BY org_id, env, customer_id, event_name, entity_id, internal_product_id, hour
 
 TYPE materialized
 DATASOURCE events_hourly_no_properties_two_mv

server/tinybird/pipes/aggregate_groupable.pipe

@@ -15,7 +15,7 @@ DESCRIPTION >
 SQL >
     %
     {% set no_property_filters = (not defined(filter_key_0) or String(filter_key_0, '') == '') and (not defined(filter_key_1) or String(filter_key_1, '') == '') and (not defined(filter_key_2) or String(filter_key_2, '') == '') and (not defined(filter_key_3) or String(filter_key_3, '') == '') and (not defined(filter_key_4) or String(filter_key_4, '') == '') %}
-    {% set use_no_props = (String(group_column, 'property') in ['customer_id', 'entity_id']) and no_property_filters %}
+    {% set use_no_props = (String(group_column, 'property') in ['customer_id', 'entity_id', 'plan_id']) and no_property_filters %}
 
     SELECT
         {% if String(bin_size, 'day') == 'hour' %}
@@ -30,6 +30,8 @@ SQL >
         customer_id as group_value,
         {% elif String(group_column, 'property') == 'entity_id' %}
         entity_id as group_value,
+        {% elif String(group_column, 'property') == 'plan_id' %}
+        coalesce(internal_product_id, '') as group_value,
         {% else %}
         {{ column('properties.' + String(property_key, '')) }}::String as group_value,
         {% end %}

shared/api/events/aggregate/eventsAggregateParams.ts

@@ -23,16 +23,17 @@ export const ExtEventsAggregateParamsSchema = z.object({
 			(val) =>
 				val.startsWith("properties.") ||
 				val === "$customer_id" ||
-				val === "$entity_id",
+				val === "$entity_id" ||
+				val === "$plan_id",
 			{
 				message:
-					'group_by must start with "properties." or be "$customer_id" or "$entity_id"',
+					'group_by must start with "properties." or be "$customer_id", "$entity_id", or "$plan_id"',
 			},
 		)
 		.optional()
 		.meta({
 			description:
-				'Property to group events by (e.g. "properties.region"), or "$customer_id" / "$entity_id" to group by those columns',
+				'Property to group events by (e.g. "properties.region"), or "$customer_id" / "$entity_id" / "$plan_id" to group by those columns',
 		}),
 	range: RangeEnum.optional().meta({
 		description: