refactor(infra): colocate stripe lake table

infra/console.ts

@@ -103,6 +103,92 @@ export const stripeWebhook = new stripe.WebhookEndpoint("StripeWebhookEndpoint",
   ],
 })
 
+export const stripeEvent = lake
+  ? (() => {
+      const billingNamespace = new aws.s3tables.Namespace("LakeBillingNamespace", {
+        namespace: "billing",
+        tableBucketArn: lake.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" },
+      )
+
+      return new sst.Linkable("StripeEvent", {
+        properties: {
+          region: lake.lakeRegion,
+          catalog: lake.lakeCatalog,
+          database: billingNamespace.namespace,
+          table: stripeTable.name,
+          tableBucket: lake.tableBucket.name,
+          workgroup: lake.lakeAthenaWorkgroup.name,
+        },
+      })
+    })()
+  : undefined
+
 const zenLiteProduct = new stripe.Product("ZenLite", {
   name: "OpenCode Go",
 })
@@ -267,6 +353,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,

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<string, unknown> | undefined {
+  if (!value || typeof value !== "object" || Array.isArray(value)) return undefined
+  return value as Record<string, unknown>
+}
+
+function array(value: unknown): Record<string, unknown>[] {
+  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<string, unknown> | undefined) {
+  return value && Object.keys(value).length > 0 ? JSON.stringify(value) : undefined
+}