From 722d0f464a0283c51685c91c7a5534ea54cb7402 Mon Sep 17 00:00:00 2001 From: amianthus <49116958+SirTenzin@users.noreply.github.com> Date: Fri, 15 May 2026 12:58:30 +0100 Subject: [PATCH] =?UTF-8?q?feat:=20=F0=9F=8E=B8=20plan=5Fid=20in=20tb=20sc?= =?UTF-8?q?hemas?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../internal/balances/track/v3/runPostgresTrackV3.ts | 7 +++++++ .../src/internal/balances/track/v3/runRedisTrackV3.ts | 11 ++++++++++- .../events/handlers/handleExternalAggregateEvents.ts | 4 ++-- .../firecrawl_backfill_apikeyid.connection | 4 ---- server/tinybird/datasources/events.datasource | 1 + .../events_by_timestamp_mv.datasource | 1 + .../materializations/events_by_timestamp_mv_pipe.pipe | 1 + .../materializations/events_hourly_mv.datasource | 1 + .../materializations/events_hourly_mv_pipe.pipe | 3 ++- .../events_hourly_no_properties_two_mv.datasource | 1 + .../events_hourly_no_properties_two_mv_pipe.pipe | 3 ++- server/tinybird/pipes/aggregate_groupable.pipe | 4 +++- shared/api/events/aggregate/eventsAggregateParams.ts | 7 ++++--- 13 files changed, 35 insertions(+), 13 deletions(-) delete mode 100644 server/tinybird/connections/firecrawl_backfill_apikeyid.connection diff --git a/server/src/internal/balances/track/v3/runPostgresTrackV3.ts b/server/src/internal/balances/track/v3/runPostgresTrackV3.ts index 8f1c0ce97..8ea43cbee 100644 --- a/server/src/internal/balances/track/v3/runPostgresTrackV3.ts +++ b/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, }); diff --git a/server/src/internal/balances/track/v3/runRedisTrackV3.ts b/server/src/internal/balances/track/v3/runRedisTrackV3.ts index 74fc2bd91..cc8b91efc 100644 --- a/server/src/internal/balances/track/v3/runRedisTrackV3.ts +++ b/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, diff --git a/server/src/internal/events/handlers/handleExternalAggregateEvents.ts b/server/src/internal/events/handlers/handleExternalAggregateEvents.ts index fb9b1051f..e404c4b90 100644 --- a/server/src/internal/events/handlers/handleExternalAggregateEvents.ts +++ b/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([ diff --git a/server/tinybird/connections/firecrawl_backfill_apikeyid.connection b/server/tinybird/connections/firecrawl_backfill_apikeyid.connection deleted file mode 100644 index 2e0c6bdf3..000000000 --- a/server/tinybird/connections/firecrawl_backfill_apikeyid.connection +++ /dev/null @@ -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 diff --git a/server/tinybird/datasources/events.datasource b/server/tinybird/datasources/events.datasource index b948c0f50..b5664af83 100644 --- a/server/tinybird/datasources/events.datasource +++ b/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` diff --git a/server/tinybird/materializations/events_by_timestamp_mv.datasource b/server/tinybird/materializations/events_by_timestamp_mv.datasource index f11a642a6..64870b017 100644 --- a/server/tinybird/materializations/events_by_timestamp_mv.datasource +++ b/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" diff --git a/server/tinybird/materializations/events_by_timestamp_mv_pipe.pipe b/server/tinybird/materializations/events_by_timestamp_mv_pipe.pipe index c85a17ad1..3b9ac4075 100644 --- a/server/tinybird/materializations/events_by_timestamp_mv_pipe.pipe +++ b/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 diff --git a/server/tinybird/materializations/events_hourly_mv.datasource b/server/tinybird/materializations/events_hourly_mv.datasource index 9ad2ae3d8..40a99998c 100644 --- a/server/tinybird/materializations/events_hourly_mv.datasource +++ b/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, diff --git a/server/tinybird/materializations/events_hourly_mv_pipe.pipe b/server/tinybird/materializations/events_hourly_mv_pipe.pipe index 92868a870..021ee3181 100644 --- a/server/tinybird/materializations/events_hourly_mv_pipe.pipe +++ b/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 diff --git a/server/tinybird/materializations/events_hourly_no_properties_two_mv.datasource b/server/tinybird/materializations/events_hourly_no_properties_two_mv.datasource index 1f8ab4ec5..afa5634de 100644 --- a/server/tinybird/materializations/events_hourly_no_properties_two_mv.datasource +++ b/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 diff --git a/server/tinybird/materializations/events_hourly_no_properties_two_mv_pipe.pipe b/server/tinybird/materializations/events_hourly_no_properties_two_mv_pipe.pipe index 8bd9bf5c2..efd97d988 100644 --- a/server/tinybird/materializations/events_hourly_no_properties_two_mv_pipe.pipe +++ b/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 diff --git a/server/tinybird/pipes/aggregate_groupable.pipe b/server/tinybird/pipes/aggregate_groupable.pipe index 4e3da59f5..c69d2bd5e 100644 --- a/server/tinybird/pipes/aggregate_groupable.pipe +++ b/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 %} diff --git a/shared/api/events/aggregate/eventsAggregateParams.ts b/shared/api/events/aggregate/eventsAggregateParams.ts index b617cafc0..6c558caf8 100644 --- a/shared/api/events/aggregate/eventsAggregateParams.ts +++ b/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: