mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-05 05:20:19 +00:00
feat(console): ingest stripe webhooks into lake
This commit is contained in:
opencode-agent[bot]
parent
b93963e462
commit
e6691facd7
3 changed files with 228 additions and 1 deletions
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
////////////////
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue