From e6691facd702af6433e8607a301a458b781d9c00 Mon Sep 17 00:00:00 2001 From: "opencode-agent[bot]" Date: Tue, 2 Jun 2026 19:57:28 +0000 Subject: [PATCH] feat(console): ingest stripe webhooks into lake --- infra/console.ts | 1 + infra/stats.ts | 82 ++++++++++ .../console/app/src/routes/stripe/webhook.ts | 146 +++++++++++++++++- 3 files changed, 228 insertions(+), 1 deletion(-) diff --git a/infra/console.ts b/infra/console.ts index 0a304a7be30..db2356a729c 100644 --- a/infra/console.ts +++ b/infra/console.ts @@ -267,6 +267,7 @@ new sst.cloudflare.x.SolidStart("Console", { SALESFORCE_INSTANCE_URL, ZEN_BLACK_PRICE, ZEN_LITE_PRICE, + ...(lake?.lakeIngest ? [lake.lakeIngest] : []), new sst.Secret("ZEN_LIMITS"), new sst.Secret("ZEN_SESSION_SECRET"), ...ZEN_MODELS, diff --git a/infra/stats.ts b/infra/stats.ts index 59dbb8d8a23..5f2259944af 100644 --- a/infra/stats.ts +++ b/infra/stats.ts @@ -87,6 +87,77 @@ const inferenceEventTable = new aws.s3tables.Table( { deleteBeforeReplace: $app.stage !== "production" }, ) +const billingNamespace = new aws.s3tables.Namespace("LakeBillingNamespace", { + namespace: "billing", + tableBucketArn: tableBucket.arn, +}) + +const stripeTable = new aws.s3tables.Table( + "LakeBillingStripeTable", + { + name: "stripe", + namespace: billingNamespace.namespace, + tableBucketArn: billingNamespace.tableBucketArn, + format: "ICEBERG", + metadata: { + iceberg: { + schema: { + fields: [ + { name: "event_timestamp", type: "string", required: false }, + { name: "event_date", type: "string", required: false }, + { name: "event_id", type: "string", required: false }, + { name: "event_type", type: "string", required: false }, + { name: "api_version", type: "string", required: false }, + { name: "livemode", type: "boolean", required: false }, + { name: "pending_webhooks", type: "int", required: false }, + { name: "request_id", type: "string", required: false }, + { name: "idempotency_key", type: "string", required: false }, + { name: "object_id", type: "string", required: false }, + { name: "object_type", type: "string", required: false }, + { name: "object_created_timestamp", type: "string", required: false }, + { name: "customer_id", type: "string", required: false }, + { name: "customer_email", type: "string", required: false }, + { name: "customer_name", type: "string", required: false }, + { name: "customer_phone", type: "string", required: false }, + { name: "workspace_id", type: "string", required: false }, + { name: "user_id", type: "string", required: false }, + { name: "user_email", type: "string", required: false }, + { name: "subscription_id", type: "string", required: false }, + { name: "invoice_id", type: "string", required: false }, + { name: "charge_id", type: "string", required: false }, + { name: "payment_intent_id", type: "string", required: false }, + { name: "payment_method_id", type: "string", required: false }, + { name: "checkout_session_id", type: "string", required: false }, + { name: "refund_id", type: "string", required: false }, + { name: "product_id", type: "string", required: false }, + { name: "price_id", type: "string", required: false }, + { name: "quantity", type: "long", required: false }, + { name: "status", type: "string", required: false }, + { name: "billing_reason", type: "string", required: false }, + { name: "collection_method", type: "string", required: false }, + { name: "cancel_at_period_end", type: "boolean", required: false }, + { name: "canceled_at_timestamp", type: "string", required: false }, + { name: "cancel_at_timestamp", type: "string", required: false }, + { name: "cancellation_reason", type: "string", required: false }, + { name: "currency", type: "string", required: false }, + { name: "amount", type: "long", required: false }, + { name: "amount_paid", type: "long", required: false }, + { name: "amount_due", type: "long", required: false }, + { name: "amount_refunded", type: "long", required: false }, + { name: "refunded", type: "boolean", required: false }, + { name: "refund_reason", type: "string", required: false }, + { name: "current_period_start_timestamp", type: "string", required: false }, + { name: "current_period_end_timestamp", type: "string", required: false }, + { name: "metadata", type: "string", required: false }, + { name: "previous_attributes", type: "string", required: false }, + ], + }, + }, + }, + }, + { deleteBeforeReplace: $app.stage !== "production" }, +) + export const inferenceEvent = new sst.Linkable("InferenceEvent", { properties: { region: lakeRegion, @@ -98,6 +169,17 @@ export const inferenceEvent = new sst.Linkable("InferenceEvent", { }, }) +export const stripeEvent = new sst.Linkable("StripeEvent", { + properties: { + region: lakeRegion, + catalog: lakeCatalog, + database: billingNamespace.namespace, + table: stripeTable.name, + tableBucket: tableBucket.name, + workgroup: lakeAthenaWorkgroup.name, + }, +}) + //////////////// // DATABASE //////////////// diff --git a/packages/console/app/src/routes/stripe/webhook.ts b/packages/console/app/src/routes/stripe/webhook.ts index dfff3bd809b..f07dfff6e36 100644 --- a/packages/console/app/src/routes/stripe/webhook.ts +++ b/packages/console/app/src/routes/stripe/webhook.ts @@ -6,7 +6,7 @@ import { BillingTable, LiteTable, PaymentTable } from "@opencode-ai/console-core import { Identifier } from "@opencode-ai/console-core/identifier.js" import { centsToMicroCents } from "@opencode-ai/console-core/util/price.js" import { Actor } from "@opencode-ai/console-core/actor.js" -import { Resource } from "@opencode-ai/console-resource" +import { Resource, waitUntil } from "@opencode-ai/console-resource" import { LiteData } from "@opencode-ai/console-core/lite.js" import { BlackData } from "@opencode-ai/console-core/black.js" import { Referral } from "@opencode-ai/console-core/referral.js" @@ -372,9 +372,153 @@ export async function POST(input: APIEvent) { } })() .then((message) => { + waitUntil( + writeStripeEventToLake(body).catch((error) => { + console.error("Stripe lake ingest failed", error) + }), + ) return Response.json({ message: message ?? "done" }, { status: 200 }) }) .catch((error: any) => { return Response.json({ message: error.message }, { status: 500 }) }) } + +async function writeStripeEventToLake(event: Stripe.Event) { + const lakeIngest = getLakeIngest() + if (!lakeIngest) return + + const response = await fetch(lakeIngest.url, { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: `Bearer ${lakeIngest.secret}`, + }, + body: JSON.stringify({ events: [toLakeStripeEvent(event)] }), + }) + if (response.ok) return + throw new Error(`Lake ingest rejected Stripe event ${event.id}: ${response.status} ${await response.text()}`) +} + +function getLakeIngest(): { url: string; secret: string } | undefined { + try { + return Resource.LakeIngest + } catch { + return undefined + } +} + +function toLakeStripeEvent(event: Stripe.Event) { + const object = record(event.data.object) ?? {} + const objectType = string(object.object) + const objectID = string(object.id) + const metadata = record(object.metadata) + const cancellationDetails = record(object.cancellation_details) + const firstLine = array(record(object.lines)?.data)[0] + const firstItem = array(record(object.items)?.data)[0] + const firstRefund = array(record(object.refunds)?.data)[0] + const firstPayment = record(array(record(object.payments)?.data)[0]?.payment) + const price = record(firstItem?.price) + const linePricing = record(firstLine?.pricing) + const linePriceDetails = record(linePricing?.price_details) + const eventTimestamp = new Date(event.created * 1000).toISOString() + + return { + _datalake_key: "billing.stripe", + event_timestamp: eventTimestamp, + event_date: eventTimestamp.slice(0, 10), + event_id: event.id, + event_type: event.type, + api_version: event.api_version, + livemode: event.livemode, + pending_webhooks: event.pending_webhooks, + request_id: event.request?.id, + idempotency_key: event.request?.idempotency_key, + object_id: objectID, + object_type: objectType, + object_created_timestamp: timestamp(integer(object.created)), + customer_id: id(object.customer) ?? (objectType === "customer" ? objectID : undefined), + customer_email: string(object.customer_email) ?? string(object.email), + customer_name: string(object.customer_name) ?? string(object.name), + customer_phone: string(object.customer_phone) ?? string(object.phone), + workspace_id: string(metadata?.workspaceID), + user_id: string(metadata?.userID), + user_email: string(metadata?.userEmail), + subscription_id: + id(object.subscription) ?? + id(record(object.parent)?.subscription_details, "subscription") ?? + (objectType === "subscription" ? objectID : undefined), + invoice_id: id(object.invoice) ?? id(object.latest_invoice) ?? (objectType === "invoice" ? objectID : undefined), + charge_id: objectType === "charge" ? objectID : id(object.charge), + payment_intent_id: id(object.payment_intent) ?? id(firstPayment, "payment_intent"), + payment_method_id: id(object.payment_method) ?? id(object.default_payment_method), + checkout_session_id: objectType === "checkout.session" ? objectID : undefined, + refund_id: objectType === "refund" ? objectID : id(firstRefund), + product_id: id(price?.product) ?? string(linePriceDetails?.product), + price_id: id(price) ?? string(linePriceDetails?.price), + quantity: integer(firstItem?.quantity) ?? integer(firstLine?.quantity), + status: string(object.status), + billing_reason: string(object.billing_reason), + collection_method: string(object.collection_method), + cancel_at_period_end: boolean(object.cancel_at_period_end), + canceled_at_timestamp: timestamp(integer(object.canceled_at)), + cancel_at_timestamp: timestamp(integer(object.cancel_at)), + cancellation_reason: string(cancellationDetails?.reason), + currency: string(object.currency), + amount: integer(object.amount), + amount_paid: integer(object.amount_paid), + amount_due: integer(object.amount_due), + amount_refunded: integer(object.amount_refunded), + refunded: boolean(object.refunded), + refund_reason: string(object.reason), + current_period_start_timestamp: timestamp(integer(object.current_period_start)), + current_period_end_timestamp: timestamp(integer(object.current_period_end)), + metadata: json(metadata), + previous_attributes: json(record(event.data.previous_attributes)), + } +} + +function record(value: unknown): Record | undefined { + if (!value || typeof value !== "object" || Array.isArray(value)) return undefined + return value as Record +} + +function array(value: unknown): Record[] { + if (!Array.isArray(value)) return [] + return value.flatMap((item) => { + const itemRecord = record(item) + return itemRecord ? [itemRecord] : [] + }) +} + +function id(value: unknown, key = "id") { + if (typeof value === "string") return value + return string(record(value)?.[key]) +} + +function string(value: unknown) { + if (typeof value === "string") return value + if (typeof value === "number" || typeof value === "boolean") return String(value) + return undefined +} + +function boolean(value: unknown) { + if (typeof value === "boolean") return value + if (typeof value === "string") return value === "true" ? true : value === "false" ? false : undefined + return undefined +} + +function integer(value: unknown) { + if (typeof value === "number" && Number.isFinite(value)) return Math.round(value) + if (typeof value !== "string") return undefined + const parsed = Number(value) + return Number.isFinite(parsed) ? Math.round(parsed) : undefined +} + +function timestamp(value: number | undefined) { + return value === undefined ? undefined : new Date(value * 1000).toISOString() +} + +function json(value: Record | undefined) { + return value && Object.keys(value).length > 0 ? JSON.stringify(value) : undefined +}