import type { ExpressionBuilder, Selectable } from 'kysely' import { sql } from 'kysely' import { Value } from 'ox' import * as Fees from '../../internal/Fees.js' import * as Id from '../../internal/Id.js' import type * as Db from '../Db.js' import type * as db_Schema from '../Schema.js' import * as Projects from './projects.js' /** Columns of the `sponsored_transactions` table, derived from `Schema.SponsoredTransaction`. */ export type Table = db_Schema.SponsoredTransaction /** A stored sponsored-transaction row. */ export type Record = Selectable /** How long failed fill intents remain eligible for on-chain reconciliation. */ export const failedIntentRecoveryTtlMs = 86_400_000 /** * Records one sponsorship commitment as `pending`, keyed by its fee-payer * sign payload. Written after fee-payer signing and before any broadcast, so * no sponsored transaction goes unrecorded; a later recording site for the * same envelope (raw submission after a fill intent) upgrades the row with * its transaction hash, reopening failed rows for finalization. * * @param db - The database. * @param input - Attribution and envelope facts captured at signing time. * @returns The stored record. */ export async function upsert(db: Db.Db, input: upsert.Input): Promise { const record: Record = { apiKeyId: input.apiKeyId, billable: input.billable, chainId: input.chainId, createdAt: new Date().toISOString(), currency: input.currency ?? null, environment: input.environment, feeAmount: null, feeMax: input.feeMax ?? null, feeToken: input.feeToken ?? null, finalizedAt: null, id: Id.generate('stx'), meterReportedAt: null, orgId: input.orgId, projectId: input.projectId ?? null, signPayload: input.signPayload, status: 'pending', transaction: input.transaction, transactionHash: input.transactionHash ?? null, } const stored = await db.kysely .insertInto('sponsored_transactions') .values(record) .onConflict((oc) => oc .column('signPayload') .doUpdateSet((eb) => ({ // Same sign payload = same envelope = same cap; only fills older nulls. currency: eb.fn.coalesce(eb.ref('sponsored_transactions.currency'), eb.ref('excluded.currency')), // prettier-ignore feeMax: eb.fn.coalesce( eb.ref('sponsored_transactions.feeMax'), eb.ref('excluded.feeMax'), ), // A raw submission fills the intent's hash; a duplicate intent never // clears an already-known hash. transactionHash: eb.fn.coalesce( eb.ref('excluded.transactionHash'), eb.ref('sponsored_transactions.transactionHash'), ), createdAt: sql`CASE WHEN ${eb.ref('sponsored_transactions.status')} = 'failed' THEN ${eb.ref('excluded.createdAt')} ELSE ${eb.ref('sponsored_transactions.createdAt')} END`, finalizedAt: sql`CASE WHEN ${eb.ref('sponsored_transactions.status')} = 'failed' THEN NULL ELSE ${eb.ref('sponsored_transactions.finalizedAt')} END`, status: sql`CASE WHEN ${eb.ref('sponsored_transactions.status')} = 'failed' THEN 'pending' ELSE ${eb.ref('sponsored_transactions.status')} END`, })) // A shared envelope cannot transfer billing attribution between owners. .where('sponsored_transactions.orgId', '=', input.orgId) .where( sql`sponsored_transactions.project_id IS NOT DISTINCT FROM ${record.projectId}`, ), ) .returningAll() .executeTakeFirst() if (!stored) throw new AttributionConflictError() return stored } export declare namespace upsert { /** Attribution and envelope facts recorded at sponsorship time. */ type Input = { /** API key id (`key_…`) that requested sponsorship. */ apiKeyId: string /** Whether the sponsorship accrues billable spend; false for sandbox. */ billable: boolean /** Chain the sponsored transaction targets. */ chainId: number /** Lowercase fee currency snapshot (`usd`), when resolved. */ currency?: string | undefined /** Key environment the sponsorship was requested under. */ environment: Table['environment'] /** Signed fee cap (`gas × maxFeePerGas`) in base units, when known. */ feeMax?: string | undefined /** Fee token the sponsorship resolved, when known. */ feeToken?: string | undefined /** Organization id (`org_…`) the spend attributes to. */ orgId: string /** Project id (`prj_…`) the spend attributes to; omit for organization-level spend. */ projectId?: string | undefined /** Fee-payer sign payload — a stable identity for the sponsored envelope. */ signPayload: string /** Serialized sponsored transaction, kept for observability beyond log retention. */ transaction: string /** Transaction hash; omit for fill intents, whose senders have not signed yet. */ transactionHash?: string | undefined } } /** * Reads one sponsorship row by id. * * @param db - The database. * @param id - The sponsored-transaction id (`stx_…`). * @returns The record, or `undefined` when absent. */ export function get(db: Db.Db, id: string): Promise { return db.kysely .selectFrom('sponsored_transactions') .selectAll() .where('id', '=', id) .executeTakeFirst() } /** * Lists pending sponsorships, oldest first, for the finalization job. * * @param db - The database. * @param options - Paging options. * @returns The pending records. */ export function listPending(db: Db.Db, options: listPending.Options = {}): Promise { let query = db.kysely .selectFrom('sponsored_transactions') .selectAll() .where('status', '=', 'pending') // `id` breaks same-millisecond `createdAt` ties for a deterministic scan. .orderBy('createdAt', 'asc') .orderBy('id', 'asc') if (options.limit !== undefined) query = query.limit(options.limit) return query.execute() } export declare namespace listPending { /** Options for {@link listPending}. */ type Options = { /** Maximum rows to return. */ limit?: number | undefined } } /** * Lists hash-less fill intents, oldest first, for the reconciliation pass. * Failed intents are included only when the caller supplies a recovery floor. * * @param db - The database. * @param options - Paging options. * @returns The pending intents. */ export function listIntents(db: Db.Db, options: listIntents.Options = {}): Promise { let query = db.kysely .selectFrom('sponsored_transactions') .selectAll() .where(intentStatus(options.failedSince)) .where('transactionHash', 'is', null) // `pending` sorts after `failed`; descending keeps active intents first. .orderBy('status', 'desc') .orderBy('createdAt', 'asc') .orderBy('id', 'asc') if (options.limit !== undefined) query = query.limit(options.limit) return query.execute() } export declare namespace listIntents { /** Options for {@link listIntents}. */ type Options = listPending.Options & { /** Include failed intents whose terminal timestamp is at or after this ISO timestamp. */ failedSince?: string | undefined } } /** * Fills a reconciled intent's transaction hash and reopens failed intents. * A no-op once another pass already assigned the hash. * * @param db - The database. * @param id - The sponsored-transaction id (`stx_…`). * @param transactionHash - The matched on-chain transaction hash. * @param limit - The active period spend limit, when enforced. */ export async function assignTransactionHash( db: Db.Db, id: string, transactionHash: string, limit?: reserve.Limit, ): Promise { try { await db.transaction(async (tx) => { const record = await get(tx, id) if ( !record || record.transactionHash !== null || !['failed', 'pending'].includes(record.status) ) return const failedSince = new Date(Date.now() - failedIntentRecoveryTtlMs).toISOString() if ( record.status === 'failed' && (record.finalizedAt === null || record.finalizedAt < failedSince) ) return const limited = limit !== undefined && (limit.billable === undefined ? record.billable : record.billable === limit.billable) if (record.status === 'failed' && limited) { const scope = limit.projectId ? `${record.orgId}:${limit.projectId}` : record.orgId await sql`select pg_advisory_xact_lock(${billingLockClass}, hashtext(${scope}))`.execute(tx.kysely) // prettier-ignore } let query = tx.kysely .updateTable('sponsored_transactions') .set({ ...(record.status === 'failed' ? { createdAt: new Date().toISOString() } : {}), finalizedAt: null, status: 'pending', transactionHash, }) .where('id', '=', id) .where('status', '=', record.status) .where('transactionHash', 'is', null) if (record.status === 'failed') query = query.where((eb) => eb.exists( eb .selectFrom('organizations') .select('organizations.id') .whereRef('organizations.id', '=', 'sponsored_transactions.orgId'), ), ) const updated = await query.executeTakeFirst() if (updated.numUpdatedRows === 0n) return if (record.status === 'failed' && limited) { const committed = await spend(tx, { ...(limit.billable === undefined ? {} : { billable: limit.billable }), chainIds: limit.chainIds, orgId: record.orgId, ...(limit.projectId === undefined ? {} : { projectId: limit.projectId }), since: limit.since, }) if (committed > limit.max) throw new PeriodSpendLimitError() } }) } catch (error) { // Keep the on-chain match eligible until budget becomes available. if (error instanceof PeriodSpendLimitError) await db.kysely .updateTable('sponsored_transactions') .set({ finalizedAt: new Date().toISOString() }) .where('id', '=', id) .where('status', '=', 'failed') .where('transactionHash', 'is', null) .execute() throw error } } /** * Transitions a pending sponsorship to `finalized` with the receipt's actual * fee. A no-op when the row is no longer pending (idempotent re-runs). * * @param db - The database. * @param id - The sponsored-transaction id (`stx_…`). * @param input - The finalized fee facts. */ export async function finalize(db: Db.Db, id: string, input: finalize.Input): Promise { await db.kysely .updateTable('sponsored_transactions') .set({ feeAmount: input.feeAmount, finalizedAt: input.finalizedAt, status: 'finalized' }) .where('id', '=', id) .where('status', '=', 'pending') .execute() } export declare namespace finalize { /** Fee facts resolved from the transaction receipt. */ type Input = { /** Actual fee paid in fee-token base units. */ feeAmount: string /** When the sponsorship finalized (ISO 8601). */ finalizedAt: string } } /** * Transitions a pending sponsorship to `failed` (never landed within the * pending TTL). A no-op when the row is no longer pending. * * @param db - The database. * @param id - The sponsored-transaction id (`stx_…`). * @param finalizedAt - When the sponsorship was marked failed (ISO 8601). */ export async function fail(db: Db.Db, id: string, finalizedAt: string): Promise { await db.kysely .updateTable('sponsored_transactions') .set({ finalizedAt, status: 'failed' }) .where('id', '=', id) .where('status', '=', 'pending') .execute() } /** * Records a sponsorship and enforces the period spend limit atomically. A * per-org advisory lock serializes concurrent sponsorships so each one's * signed cap is counted before the limit check; without it a burst reads * pre-burst spend and overshoots. Throws {@link PeriodSpendLimitError} (rolling * back the row) when the limit would be exceeded. * * @param db - The database (primary; never a cached replica). * @param input - The sponsorship to record. * @param limit - The period spend limit to enforce. * @returns The stored record. */ export async function reserve( db: Db.Db, input: upsert.Input, limit: reserve.Limit, ): Promise { return db.transaction(async (tx) => { // `hashtext` maps the org id into the advisory-lock keyspace; the class // namespaces it against other advisory-lock users. Released at tx end. const scope = limit.projectId ? `${input.orgId}:${limit.projectId}` : input.orgId await sql`select pg_advisory_xact_lock(${billingLockClass}, hashtext(${scope}))`.execute(tx.kysely) // prettier-ignore const record = await upsert(tx, input) const limited = limit.billable === undefined ? record.billable : record.billable === limit.billable if (limited) { const committed = await spend(tx, { ...(limit.billable === undefined ? {} : { billable: limit.billable }), chainIds: limit.chainIds, orgId: input.orgId, ...(limit.projectId === undefined ? {} : { projectId: limit.projectId }), since: limit.since, }) if (committed > limit.max) throw new PeriodSpendLimitError() } return record }) } /** Advisory-lock class namespacing per-org sponsorship serialization. */ const billingLockClass = 1112294220 export declare namespace reserve { /** The period spend limit enforced by {@link reserve}. */ type Limit = { /** Billing snapshot whose rows consume this limit; omit for customer-billable spend. */ billable?: boolean | undefined /** Chains whose spend counts (mainnet chain ids). */ chainIds: readonly number[] /** Limit in fee-token base units. */ max: bigint /** Project scope for independent limits; omit for the organization-wide budget. */ projectId?: string | undefined /** Window start (ISO 8601). */ since: string } } /** * Applies the current promotion policy and records the sponsorship atomically. * Exhausted promotions retain their window and fall back to customer billing. */ export function reservePromotion( db: Db.Db, input: upsert.Input, options: reservePromotion.Options, ): Promise { return db.transaction(async (tx) => { const scope = `${input.orgId}:${options.projectId}` await sql`select pg_advisory_xact_lock(${billingLockClass}, hashtext(${scope}))`.execute(tx.kysely) // prettier-ignore const existing = await tx.kysely .selectFrom('sponsored_transactions') .selectAll() .where('signPayload', '=', input.signPayload) .forUpdate() .executeTakeFirst() if (existing) { if (existing.orgId !== input.orgId || existing.projectId !== options.projectId) throw new AttributionConflictError() const stored = await upsert(tx, input) if (stored.billable && options.billingLimit) { await sql`select pg_advisory_xact_lock(${billingLockClass}, hashtext(${input.orgId}))`.execute(tx.kysely) // prettier-ignore const billed = await spend(tx, { chainIds: options.billingLimit.chainIds, orgId: input.orgId, since: options.billingLimit.since, }) if (billed > options.billingLimit.max) throw new PeriodSpendLimitError() } if (!stored.billable) { const project = await Projects.get(tx, options.projectId) if (!project?.sponsorshipSubsidyStartsAt) throw new PromotionActivationError() if (project.sponsorshipSpendLimit !== null) { const committed = await spend(tx, { billable: false, chainIds: options.chainIds, orgId: input.orgId, projectId: options.projectId, since: project.sponsorshipSubsidyStartsAt, }) if (committed > BigInt(project.sponsorshipSpendLimit)) throw new PeriodSpendLimitError() } } return { record: stored, status: stored.billable ? 'billed' : 'subsidized' } } const organization = await tx.kysely .selectFrom('organizations') .select(['sponsorshipSubsidyDurationDays', 'sponsorshipSubsidyProjectSpendLimit']) .where('id', '=', input.orgId) .forShare() .executeTakeFirst() if (!organization || organization.sponsorshipSubsidyDurationDays === null) return { status: 'ineligible' } const current = await Projects.get(tx, options.projectId) if (!current || current.orgId !== input.orgId) throw new PromotionActivationError() if ( current.sponsorshipSubsidyEndsAt !== null && Date.parse(options.at) >= Date.parse(current.sponsorshipSubsidyEndsAt) ) return { status: 'ineligible' } const endsAt = new Date( Date.parse(options.at) + organization.sponsorshipSubsidyDurationDays * 86_400_000, ).toISOString() const project = await Projects.activateSponsorship(tx, { at: options.at, endsAt, id: options.projectId, orgId: input.orgId, spendLimit: organization.sponsorshipSubsidyProjectSpendLimit === null ? null : Value.from( organization.sponsorshipSubsidyProjectSpendLimit, Fees.tokenDecimals, ).toString(), }) if (!project?.sponsorshipSubsidyStartsAt || !project.sponsorshipSubsidyEndsAt) throw new PromotionActivationError() let record = await upsert(tx, input) if (project.sponsorshipSpendLimit !== null) { const committed = await spend(tx, { billable: false, chainIds: options.chainIds, orgId: input.orgId, projectId: options.projectId, since: project.sponsorshipSubsidyStartsAt, }) if (committed > BigInt(project.sponsorshipSpendLimit)) { record = await tx.kysely .updateTable('sponsored_transactions') .set({ billable: true }) .where('id', '=', record.id) .returningAll() .executeTakeFirstOrThrow() if (options.billingLimit) { await sql`select pg_advisory_xact_lock(${billingLockClass}, hashtext(${input.orgId}))`.execute(tx.kysely) // prettier-ignore const billed = await spend(tx, { chainIds: options.billingLimit.chainIds, orgId: input.orgId, since: options.billingLimit.since, }) if (billed > options.billingLimit.max) throw new PeriodSpendLimitError() } } } return { record, status: record.billable ? 'billed' : 'subsidized' } }) } export declare namespace reservePromotion { /** Project promotion window and cap. */ type Options = { /** Timestamp of the recorded sponsorship that may activate the window. */ at: string /** Customer billing limit used when the project promotion is exhausted. */ billingLimit?: reserve.Limit | undefined /** Mainnet chains whose promotional spend counts. */ chainIds: readonly number[] /** Project receiving the independent promotion. */ projectId: string } /** Promotion reservation outcome. */ type Result = | { /** Stored sponsorship. */ record: Record /** Whether Tempo or the customer pays for the stored sponsorship. */ status: 'billed' | 'subsidized' } | { /** The current organization or project policy does not grant a promotion. */ status: 'ineligible' } } /** * Billable spend committed since a cutoff, in fee-token base units: finalized * rows at their actual fee plus pending rows at their signed fee cap, so * in-flight sponsorships consume limit budget until receipts land. * * @param db - The database. * @param options - Aggregation scope. * @returns The committed spend in base units. */ export async function spend(db: Db.Db, options: spend.Options): Promise { const { billable, chainIds, environment = 'production', orgId, projectId, since } = options let query = db.kysely .selectFrom('sponsored_transactions') .select([ sql`coalesce(sum(case when status = 'finalized' then fee_amount::numeric else 0 end), 0)`.as('finalized'), // prettier-ignore sql`coalesce(sum(case when status = 'pending' then fee_max::numeric else 0 end), 0)`.as('pending'), // prettier-ignore ]) .where('orgId', '=', orgId) .where('environment', '=', environment) // Redundant with the CASE arms, but lets the billing index's // (org_id, status, created_at) shape serve the range scan. .where('status', 'in', ['finalized', 'pending']) .where('chainId', 'in', [...chainIds]) .where('createdAt', '>=', since) if (billable !== undefined) query = query.where('billable', '=', billable) if (projectId !== undefined) query = query.where('projectId', '=', projectId) // Production mirrors the sponsorship gate — only billable rows count. Sandbox // rows are non-billable by construction, so the figure is display-only. const row = await ( environment === 'production' && billable === undefined ? query.where('billable', '=', true) : query ).executeTakeFirstOrThrow() return BigInt(row.finalized) + BigInt(row.pending) } export declare namespace spend { /** Options for {@link spend}. */ type Options = { /** Billing snapshot to count; omit for the existing production-billable behavior. */ billable?: boolean | undefined /** Chains whose spend counts (mainnet chain ids). */ chainIds: readonly number[] /** Environment whose spend counts; production mirrors the billing gate, sandbox is display-only. Defaults to `production`. */ environment?: Table['environment'] | undefined /** Organization id (`org_…`) the spend attributes to. */ orgId: string /** Project whose spend counts; omit for organization-wide spend. */ projectId?: string | undefined /** Window start (ISO 8601); rows created earlier are out of scope. */ since: string } } /** * Sponsorship usage bucketed by time for one organization: row counts and * committed fees per bucket. Fee semantics mirror {@link spend}: finalized * rows at their actual fee, pending rows at their signed cap, failed rows at * zero. Buckets with no rows are omitted. * * @param db - The database. * @param options - Aggregation scope. * @returns Time-ordered usage buckets. */ export async function usage(db: Db.Db, options: usage.Options): Promise { // TODO: convert to an OLAP-friendly approach (rollup table or columnar store) // when per-org range scans grow slow. const { environment, from, interval, orgId, projectId, to } = options // `sql.lit` inlines the enum-validated interval so the GROUP BY expression // matches the SELECT expression textually; truncation happens in UTC wall // time regardless of the session time zone. const bucket = sql`date_trunc(${sql.lit(interval)}, (created_at::timestamptz) at time zone 'UTC')` let query = db.kysely .selectFrom('sponsored_transactions') .select([ sql`to_char(${bucket}, 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')`.as('timestamp'), sql`count(*)`.as('count'), sql`count(*) filter (where status = 'failed')`.as('failed'), sql`coalesce(sum(case when status = 'finalized' then fee_amount::numeric when status = 'pending' then fee_max::numeric else 0 end), 0)`.as('feeTotal'), // prettier-ignore ]) .where('orgId', '=', orgId) // Text-range bounds keep the (org_id, created_at) index serving the scan. .where('createdAt', '>=', from) .where('createdAt', '<', to) .groupBy(bucket) .orderBy(bucket) if (environment !== undefined) query = query.where('environment', '=', environment) if (projectId !== undefined) query = query.where('projectId', '=', projectId) const rows = await query.execute() return rows.map((row) => ({ count: Number(row.count), failed: Number(row.failed), feeTotal: BigInt(row.feeTotal), timestamp: row.timestamp, })) } export declare namespace usage { /** One time bucket of sponsorship usage. */ type Bucket = { /** Sponsored transactions recorded in the bucket, any status. */ count: number /** Sponsored transactions in the bucket whose status is `failed`. */ failed: number /** Committed fees in fee-token base units: finalized fees plus pending caps. */ feeTotal: bigint /** Bucket start (ISO 8601), aligned to UTC calendar boundaries. */ timestamp: string } /** Options for {@link usage}. */ type Options = { /** Key environment to restrict to; omit for all. */ environment?: Table['environment'] | undefined /** Window start (ISO 8601), inclusive. */ from: string /** Bucket width, aligned to UTC calendar boundaries. */ interval: 'day' | 'hour' | 'month' | 'week' /** Organization id (`org_…`) the usage attributes to. */ orgId: string /** Project id (`prj_…`) to restrict to; omit for all. */ projectId?: string | undefined /** Window end (ISO 8601), exclusive. */ to: string } } /** * Lists finalized billable rows not yet reported to the billing meter, oldest * first, for the metering job. * * @param db - The database. * @param options - Reporting scope. * @returns The unreported records. */ export function listUnreported(db: Db.Db, options: listUnreported.Options): Promise { let query = db.kysely .selectFrom('sponsored_transactions') .selectAll() .where('status', '=', 'finalized') .where('billable', '=', true) .where('chainId', 'in', [...options.chainIds]) .where('meterReportedAt', 'is', null) // `id` breaks same-millisecond `finalizedAt` ties for a deterministic scan. .orderBy('finalizedAt', 'asc') .orderBy('id', 'asc') if (options.limit !== undefined) query = query.limit(options.limit) return query.execute() } export declare namespace listUnreported { /** Options for {@link listUnreported}. */ type Options = { /** Chains whose spend is metered (mainnet chain ids). */ chainIds: readonly number[] /** Maximum rows to return. */ limit?: number | undefined } } /** * Marks a row as reported to the billing meter. A no-op once marked — the * durable exactly-once guard for the metering job. * * @param db - The database. * @param id - The sponsored-transaction id (`stx_…`). * @param at - When the report was acknowledged (ISO 8601). */ export async function markReported(db: Db.Db, id: string, at: string): Promise { await db.kysely .updateTable('sponsored_transactions') .set({ meterReportedAt: at }) .where('id', '=', id) .where('meterReportedAt', 'is', null) .execute() } /** Intent status predicate; failed rows are recovery candidates only briefly. */ function intentStatus(failedSince: string | undefined) { return (eb: ExpressionBuilder) => { if (failedSince === undefined) return eb('status', '=', 'pending') return eb.or([ eb('status', '=', 'pending'), eb.and([eb('status', '=', 'failed'), eb('finalizedAt', '>=', failedSince)]), ]) } } /** Thrown by {@link reserve} when recording would exceed the org's period spend limit. */ export class PeriodSpendLimitError extends Error { override name = 'SponsoredTransactions.PeriodSpendLimitError' } /** Thrown when a promotional window cannot be activated consistently. */ export class PromotionActivationError extends Error { override name = 'SponsoredTransactions.PromotionActivationError' } /** Thrown when one signed envelope is presented with conflicting project attribution. */ export class AttributionConflictError extends Error { override name = 'SponsoredTransactions.AttributionConflictError' }