diff --git a/packages/billing/lib/clients/orb/adapters.ts b/packages/billing/lib/clients/orb/adapters.ts index bab20a2f03..2920d0731b 100644 --- a/packages/billing/lib/clients/orb/adapters.ts +++ b/packages/billing/lib/clients/orb/adapters.ts @@ -2,11 +2,20 @@ import { uuidv7 } from 'uuidv7'; import { Err, Ok } from '@nangohq/utils'; +import { envs } from '../../envs.js'; import { putOrbCustomerSchema } from './types.js'; import type { BillingAddress, BillingCustomer, BillingEvent, BillingInvoicingDetails, Result, UsageMetric } from '@nangohq/types'; import type Orb from 'orb-billing'; +// Keyed on the EVENT's timestamp, not the wall clock, so a batched or +// late-emitted event whose logical time is pre-cutover ships under the +// pre-cutover name and vice versa. See BILLING_EVENTS_CUTOVER_AT in +// packages/utils. +function cutoverAppliesTo(eventTimestamp: Date): boolean { + return !!envs.BILLING_EVENTS_CUTOVER_AT && eventTimestamp >= new Date(envs.BILLING_EVENTS_CUTOVER_AT); +} + export function toOrbEvent(event: BillingEvent): Orb.Events.EventIngestParams.Event { const { idempotencyKey, timestamp, accountId, ...rest } = event.properties; @@ -24,7 +33,7 @@ export function toOrbEvent(event: BillingEvent): Orb.Events.EventIngestParams.Ev } return { - event_name: event.type, + event_name: `${event.type}${cutoverAppliesTo(timestamp) ? '_http' : ''}`, idempotency_key: idempotencyKey || uuidv7(), external_customer_id: accountId.toString(), timestamp: timestamp.toISOString(), diff --git a/packages/billing/lib/clients/orb/adapters.unit.test.ts b/packages/billing/lib/clients/orb/adapters.unit.test.ts index be6de11117..8eb3465a1b 100644 --- a/packages/billing/lib/clients/orb/adapters.unit.test.ts +++ b/packages/billing/lib/clients/orb/adapters.unit.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, vi } from 'vitest'; +import { envs } from '../../envs.js'; import { fromOrbAddress, fromOrbCustomer, orbMetricToUsageMetric, toOrbEvent, toOrbPutCustomerPayload } from './adapters.js'; import type { BillingEvent, BillingInvoicingDetails } from '@nangohq/types'; @@ -66,6 +67,49 @@ describe('toOrbEvent', () => { expect(result.external_customer_id).toBe('42'); expect(result.timestamp).toBe('2024-01-15T10:00:00.000Z'); }); + + it('appends "_http" when the event timestamp is at or after the cutover', () => { + const originalCutover = envs.BILLING_EVENTS_CUTOVER_AT; + try { + (envs as any).BILLING_EVENTS_CUTOVER_AT = '2000-01-01T00:00:00Z'; + const event: BillingEvent = { type: 'proxy', properties: { ...baseProperties } as any }; + expect(toOrbEvent(event).event_name).toBe('proxy_http'); + } finally { + (envs as any).BILLING_EVENTS_CUTOVER_AT = originalCutover; + } + }); + + it('does not append "_http" when the event timestamp is before the cutover', () => { + const originalCutover = envs.BILLING_EVENTS_CUTOVER_AT; + try { + (envs as any).BILLING_EVENTS_CUTOVER_AT = '9999-01-01T00:00:00Z'; + const event: BillingEvent = { type: 'proxy', properties: { ...baseProperties } as any }; + expect(toOrbEvent(event).event_name).toBe('proxy'); + } finally { + (envs as any).BILLING_EVENTS_CUTOVER_AT = originalCutover; + } + }); + + it('keys the suffix on each event timestamp independently — a batched pre-cutover event stays unsuffixed even when processed after cutover', () => { + // Same cutover instant, two events on either side of it: verifies the + // suffix is decided per-event, not from wall-clock at processing time. + const originalCutover = envs.BILLING_EVENTS_CUTOVER_AT; + try { + (envs as any).BILLING_EVENTS_CUTOVER_AT = '2024-06-01T00:00:00Z'; + const beforeCutover: BillingEvent = { + type: 'proxy', + properties: { ...baseProperties, timestamp: new Date('2024-05-31T23:59:59.999Z') } as any + }; + const atCutover: BillingEvent = { + type: 'proxy', + properties: { ...baseProperties, timestamp: new Date('2024-06-01T00:00:00.000Z') } as any + }; + expect(toOrbEvent(beforeCutover).event_name).toBe('proxy'); + expect(toOrbEvent(atCutover).event_name).toBe('proxy_http'); + } finally { + (envs as any).BILLING_EVENTS_CUTOVER_AT = originalCutover; + } + }); }); // ─── toOrbPutCustomerPayload ────────────────────────────────────────────────── diff --git a/packages/metering/lib/crons/billingEventsS3Export.ts b/packages/metering/lib/crons/billingEventsS3Export.ts index f102139044..7edf71ddc3 100644 --- a/packages/metering/lib/crons/billingEventsS3Export.ts +++ b/packages/metering/lib/crons/billingEventsS3Export.ts @@ -17,7 +17,15 @@ const cronMinute = envs.CRON_BILLING_EVENTS_S3_HOURLY_EXPORT_MINUTE; const bucket = envs.BILLING_EVENTS_S3_BUCKET; const roleArn = envs.BILLING_EVENTS_S3_WRITER_ROLE_ARN; const region = envs.BILLING_EVENTS_S3_REGION; -const eventNameSuffix = envs.BILLING_EVENTS_S3_EVENT_NAME_SUFFIX ?? ''; + +// Keyed on the DATA DAY, not the wall clock. Matters at the seam: the +// Aug 1 00:15 UTC firing exports July 31 data, and July 31 is still +// pre-cutover — so those rows must ship as "_s3" shadow, not canonical. +// See BILLING_EVENTS_CUTOVER_AT in packages/utils. +function dayIsPostCutover(day: string): boolean { + if (!envs.BILLING_EVENTS_CUTOVER_AT) return false; + return new Date(`${day}T00:00:00Z`) >= new Date(envs.BILLING_EVENTS_CUTOVER_AT); +} const LOCK_KEY = 'lock:cron:billingEventsS3Export'; // Cron fires hourly; lock should expire well before the next tick. @@ -214,13 +222,12 @@ export function billingEventsS3ExportCron(): void { }); } -function addEventNameSuffix(eventName: MetricSpec['canonicalEventName']): string { - if (eventName == 'data_transfer') { +function addEventNameSuffix(eventName: MetricSpec['canonicalEventName'], day: string): string { + if (eventName === 'data_transfer') { // data_transfer is a new event and thus doesn't need the suffix to support live traffic cutoff. return eventName; } - - return `${eventName}${eventNameSuffix}`; + return `${eventName}${dayIsPostCutover(day) ? '' : '_s3'}`; } export async function exec(): Promise { @@ -236,7 +243,7 @@ export async function exec(): Promise { let anyFailure = false; try { for (const metric of METRICS) { - const eventName = addEventNameSuffix(metric.canonicalEventName); + const eventName = addEventNameSuffix(metric.canonicalEventName, day); const key = objectKey({ day, eventName }); const start = process.hrtime.bigint(); // Tracks which step is in flight so a catch can tag the failure diff --git a/packages/utils/lib/environment/parse.ts b/packages/utils/lib/environment/parse.ts index cd7469a088..b97143e78f 100644 --- a/packages/utils/lib/environment/parse.ts +++ b/packages/utils/lib/environment/parse.ts @@ -352,7 +352,14 @@ export const ENVS = z.object({ BILLING_INGEST_MAX_RETRY: z.coerce.number().optional().default(3), BILLING_EVENTS_S3_BUCKET: z.string().optional(), BILLING_EVENTS_S3_WRITER_ROLE_ARN: z.string().optional(), - BILLING_EVENTS_S3_EVENT_NAME_SUFFIX: z.string().optional(), + // Temporary. ISO 8601 timestamp at which the S3-fed pipeline becomes + // authoritative for billing. Before this instant, S3 events ship as + // "_s3" shadow and HTTP events ship canonical (unsuffixed). At + // and after this instant, the two swap roles — HTTP events pick up + // the "_http" suffix and S3 events become canonical. Unset (or set + // to a future date) to defer or roll back the cutover. Remove once + // the HTTP emission path is retired. + BILLING_EVENTS_CUTOVER_AT: z.string().datetime().optional(), BILLING_EVENTS_S3_REGION: z.string().optional().default('us-west-2'), // DLQ bucket Orb writes to when it can't ingest a billing event. Watched by the // metering DLQ monitor cron (CRON_BILLING_EVENTS_S3_DLQ_MONITOR_MINUTE).