From 7854223d5e7cf4e6b65e1c823c82e2aa962d5f5d Mon Sep 17 00:00:00 2001 From: Developer Date: Sun, 28 Jun 2026 06:30:43 -0700 Subject: [PATCH] Add bulk deduct endpoint --- docs/openapi.json | 404 +++++++++++-------- src/routes/billing.ts | 136 ++++--- src/routes/billing/deduct/bulk.test.ts | 265 +++++++++++++ src/routes/billing/deduct/bulk.ts | 242 ++++++++++++ src/services/billing.bulk.test.ts | 211 ++++++++++ src/services/billing.ts | 516 +++++++++++++++++++++++-- 6 files changed, 1524 insertions(+), 250 deletions(-) create mode 100644 src/routes/billing/deduct/bulk.test.ts create mode 100644 src/routes/billing/deduct/bulk.ts create mode 100644 src/services/billing.bulk.test.ts diff --git a/docs/openapi.json b/docs/openapi.json index c130e7a..af98c1e 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -109,6 +109,26 @@ } } }, + "409": { + "description": "Idempotency conflict", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "429": { + "description": "Rate limit exceeded", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, "500": { "description": "Internal server error", "content": { @@ -118,41 +138,99 @@ } } } + } + } + } + }, + "/api/billing/deduct/bulk": { + "post": { + "summary": "Deduct billing balance for up to 100 API calls in one batch", + "description": "Creates usage events for up to 100 billing entries in a single database transaction, then performs one aggregate on-chain deduction for the newly inserted entries. Existing requestIds are treated idempotently and are returned in the per-entry results without being charged again.", + "security": [ + { + "bearerAuth": [] + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/BillingBulkDeductRequest" + } + } + } + }, + "responses": { + "200": { + "description": "Bulk billing deduction successfully processed", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/BillingBulkDeductResponse" + } + } + } }, - "responses": { - "200": { - "description": "Success" - }, - "400": { - "description": "Bad Request" - }, - "401": { - "description": "Unauthorized" - }, - "402": { - "description": "Payment Required" - }, - "409": { - "description": "Idempotency conflict", - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ErrorResponse" - } + "400": { + "description": "Invalid request body parameters", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" } } - }, - "429": { - "description": "Rate limit exceeded", - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ErrorResponse" - } + } + }, + "401": { + "description": "Unauthorized", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" } } - }, - "500": {} + } + }, + "402": { + "description": "Payment required (insufficient aggregate balance)", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "409": { + "description": "Idempotency conflict", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "429": { + "description": "Rate limit exceeded", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "500": { + "description": "Internal server error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } } } } @@ -223,11 +301,7 @@ "required": false, "schema": { "type": "string", - "enum": [ - "day", - "week", - "month" - ] + "enum": ["day", "week", "month"] } } ], @@ -367,10 +441,7 @@ }, "type": { "type": "string", - "enum": [ - "spike", - "drop" - ] + "enum": ["spike", "drop"] }, "calls": { "type": "integer" @@ -1038,25 +1109,123 @@ } } }, - "UsageResponse": { + "BillingBulkDeductEntry": { "type": "object", "required": [ - "events", - "stats", - "period" + "requestId", + "apiId", + "endpointId", + "apiKeyId", + "amountUsdc" ], + "properties": { + "requestId": { + "type": "string" + }, + "apiId": { + "type": "string" + }, + "endpointId": { + "type": "string" + }, + "apiKeyId": { + "type": "string" + }, + "amountUsdc": { + "type": "string", + "description": "Positive decimal string with up to 7 fractional digits" + } + } + }, + "BillingBulkDeductRequest": { + "type": "object", + "required": ["entries"], + "properties": { + "entries": { + "type": "array", + "minItems": 1, + "maxItems": 100, + "items": { + "$ref": "#/components/schemas/BillingBulkDeductEntry" + } + }, + "idempotencyKey": { + "type": "string" + } + } + }, + "BillingBulkDeductEntryResult": { + "type": "object", + "required": [ + "requestId", + "usageEventId", + "alreadyProcessed", + "deductionApplied", + "reconciliationRequired" + ], + "properties": { + "requestId": { + "type": "string" + }, + "usageEventId": { + "type": "string" + }, + "stellarTxHash": { + "type": "string" + }, + "alreadyProcessed": { + "type": "boolean" + }, + "deductionApplied": { + "type": "boolean" + }, + "reconciliationRequired": { + "type": "boolean" + } + } + }, + "BillingBulkDeductResponse": { + "type": "object", + "required": [ + "success", + "entryCount", + "deductedCount", + "totalDeductedAmountUsdc", + "results" + ], + "properties": { + "success": { + "type": "boolean" + }, + "entryCount": { + "type": "integer" + }, + "deductedCount": { + "type": "integer" + }, + "totalDeductedAmountUsdc": { + "type": "string" + }, + "stellarTxHash": { + "type": "string" + }, + "results": { + "type": "array", + "items": { + "$ref": "#/components/schemas/BillingBulkDeductEntryResult" + } + } + } + }, + "UsageResponse": { + "type": "object", + "required": ["events", "stats", "period"], "properties": { "events": { "type": "array", "items": { "type": "object", - "required": [ - "id", - "apiId", - "endpoint", - "occurredAt", - "revenue" - ], + "required": ["id", "apiId", "endpoint", "occurredAt", "revenue"], "properties": { "id": { "type": "string" @@ -1079,11 +1248,7 @@ }, "stats": { "type": "object", - "required": [ - "totalCalls", - "totalSpent", - "breakdownByApi" - ], + "required": ["totalCalls", "totalSpent", "breakdownByApi"], "properties": { "totalCalls": { "type": "integer" @@ -1095,11 +1260,7 @@ "type": "array", "items": { "type": "object", - "required": [ - "apiId", - "calls", - "revenue" - ], + "required": ["apiId", "calls", "revenue"], "properties": { "apiId": { "type": "string" @@ -1117,11 +1278,7 @@ "type": "array", "items": { "type": "object", - "required": [ - "period", - "calls", - "revenue" - ], + "required": ["period", "calls", "revenue"], "properties": { "period": { "type": "string" @@ -1139,10 +1296,7 @@ }, "period": { "type": "object", - "required": [ - "from", - "to" - ], + "required": ["from", "to"], "properties": { "from": { "type": "string", @@ -1201,17 +1355,11 @@ "type": "integer" }, "firstEventAt": { - "type": [ - "string", - "null" - ], + "type": ["string", "null"], "format": "date-time" }, "lastEventAt": { - "type": [ - "string", - "null" - ], + "type": ["string", "null"], "format": "date-time" }, "statusCodes": { @@ -1224,9 +1372,7 @@ }, "AdminUsageSnapshotResponse": { "type": "object", - "required": [ - "data" - ], + "required": ["data"], "properties": { "data": { "$ref": "#/components/schemas/AdminUsageSnapshot" @@ -1235,17 +1381,11 @@ }, "AdminUsageResetResponse": { "type": "object", - "required": [ - "data" - ], + "required": ["data"], "properties": { "data": { "type": "object", - "required": [ - "developerId", - "reset", - "priorValues" - ], + "required": ["developerId", "reset", "priorValues"], "properties": { "developerId": { "type": "string" @@ -1262,10 +1402,7 @@ }, "ApisListResponse": { "type": "object", - "required": [ - "data", - "pagination" - ], + "required": ["data", "pagination"], "properties": { "data": { "type": "array", @@ -1275,11 +1412,7 @@ }, "pagination": { "type": "object", - "required": [ - "limit", - "offset", - "total" - ], + "required": ["limit", "offset", "total"], "properties": { "limit": { "type": "integer" @@ -1296,12 +1429,7 @@ }, "ApiCreateRequest": { "type": "object", - "required": [ - "name", - "base_url", - "category", - "endpoints" - ], + "required": ["name", "base_url", "category", "endpoints"], "properties": { "name": { "type": "string" @@ -1319,11 +1447,7 @@ "type": "array", "items": { "type": "object", - "required": [ - "path", - "method", - "price_per_call_usdc" - ], + "required": ["path", "method", "price_per_call_usdc"], "properties": { "path": { "type": "string" @@ -1430,9 +1554,7 @@ }, "BulkEndpointRegistrationRequest": { "type": "object", - "required": [ - "endpoints" - ], + "required": ["endpoints"], "properties": { "endpoints": { "type": "array", @@ -1440,11 +1562,7 @@ "maxItems": 50, "items": { "type": "object", - "required": [ - "path", - "method", - "price_per_call_usdc" - ], + "required": ["path", "method", "price_per_call_usdc"], "properties": { "path": { "type": "string", @@ -1477,9 +1595,7 @@ }, "BulkEndpointRegistrationResponse": { "type": "object", - "required": [ - "endpoints" - ], + "required": ["endpoints"], "properties": { "endpoints": { "type": "array", @@ -1556,10 +1672,7 @@ }, "developer": { "type": "object", - "required": [ - "id", - "name" - ], + "required": ["id", "name"], "properties": { "id": { "type": "integer" @@ -1607,19 +1720,11 @@ }, "DeveloperRevenueResponse": { "type": "object", - "required": [ - "summary", - "settlements", - "pagination" - ], + "required": ["summary", "settlements", "pagination"], "properties": { "summary": { "type": "object", - "required": [ - "total_earned", - "pending", - "available_to_withdraw" - ], + "required": ["total_earned", "pending", "available_to_withdraw"], "properties": { "total_earned": { "type": "number" @@ -1669,11 +1774,7 @@ }, "pagination": { "type": "object", - "required": [ - "limit", - "offset", - "total" - ], + "required": ["limit", "offset", "total"], "properties": { "limit": { "type": "integer" @@ -1690,21 +1791,14 @@ }, "GatewayHealthResponse": { "type": "object", - "required": [ - "apiSlug", - "latency", - "breaker" - ], + "required": ["apiSlug", "latency", "breaker"], "properties": { "apiSlug": { "type": "string" }, "latency": { "type": "object", - "required": [ - "p50", - "p95" - ], + "required": ["p50", "p95"], "properties": { "p50": { "type": "number", @@ -1720,17 +1814,11 @@ }, "breaker": { "type": "object", - "required": [ - "state" - ], + "required": ["state"], "properties": { "state": { "type": "string", - "enum": [ - "closed", - "open", - "half-open" - ] + "enum": ["closed", "open", "half-open"] } } } @@ -1738,11 +1826,7 @@ }, "ErrorResponse": { "type": "object", - "required": [ - "code", - "message", - "requestId" - ], + "required": ["code", "message", "requestId"], "properties": { "code": { "$ref": "#/components/schemas/ErrorCode" diff --git a/src/routes/billing.ts b/src/routes/billing.ts index 682b9d3..5580623 100644 --- a/src/routes/billing.ts +++ b/src/routes/billing.ts @@ -1,6 +1,6 @@ -import { Router } from 'express'; -import type { NextFunction, Request, Response } from 'express'; -import type { Pool } from 'pg'; +import { Router } from "express"; +import type { NextFunction, Request, Response } from "express"; +import type { Pool } from "pg"; import { BadGatewayError, @@ -10,19 +10,30 @@ import { NotFoundError, PaymentRequiredError, UnauthorizedError, -} from '../errors/index.js'; -import { requireAuth, type AuthenticatedLocals } from '../middleware/requireAuth.js'; -import { idempotencyMiddleware } from '../middleware/idempotency.js'; -import { billingDeductHistogramMiddleware } from '../middleware/metricsHistogram.js'; -import { BillingService, type BillingDeductResult } from '../services/billing.js'; -import { createSorobanRpcBillingClient, SorobanRpcError } from '../services/sorobanBilling.js'; -import { redactSimulationDetails } from '../lib/simulationDiagnostics.js'; -import creditsRouter from './billing/credits.js'; +} from "../errors/index.js"; +import { + requireAuth, + type AuthenticatedLocals, +} from "../middleware/requireAuth.js"; +import { idempotencyMiddleware } from "../middleware/idempotency.js"; +import { billingDeductHistogramMiddleware } from "../middleware/metricsHistogram.js"; +import { + BillingService, + type BillingDeductResult, +} from "../services/billing.js"; +import { + createSorobanRpcBillingClient, + SorobanRpcError, +} from "../services/sorobanBilling.js"; +import { redactSimulationDetails } from "../lib/simulationDiagnostics.js"; +import creditsRouter from "./billing/credits.js"; +import bulkDeductRouter from "./billing/deduct/bulk.js"; const router = Router(); -// Mount credits sub-router -router.use('/credits', creditsRouter); +// Mount billing sub-routers +router.use("/credits", creditsRouter); +router.use("/deduct", bulkDeductRouter); interface BillingDeductBody { requestId?: unknown; @@ -35,29 +46,36 @@ interface BillingDeductBody { function createRouteBillingService(pool: Pool): BillingService { const sorobanClient = createSorobanRpcBillingClient({ - rpcUrl: process.env.SOROBAN_BILLING_RPC_URL ?? process.env.SOROBAN_RPC_URL ?? 'http://localhost:8000', - contractId: process.env.SOROBAN_BILLING_CONTRACT_ID ?? 'vault_contract', + rpcUrl: + process.env.SOROBAN_BILLING_RPC_URL ?? + process.env.SOROBAN_RPC_URL ?? + "http://localhost:8000", + contractId: process.env.SOROBAN_BILLING_CONTRACT_ID ?? "vault_contract", sourceAccount: process.env.SOROBAN_BILLING_SOURCE_ACCOUNT, networkPassphrase: process.env.SOROBAN_BILLING_NETWORK_PASSPHRASE, - requestTimeoutMs: Number(process.env.SOROBAN_BILLING_RPC_TIMEOUT_MS ?? 5_000), - balanceFunctionName: process.env.SOROBAN_BILLING_BALANCE_FN ?? 'balance', - deductFunctionName: process.env.SOROBAN_BILLING_DEDUCT_FN ?? 'deduct', + requestTimeoutMs: Number( + process.env.SOROBAN_BILLING_RPC_TIMEOUT_MS ?? 5_000, + ), + balanceFunctionName: process.env.SOROBAN_BILLING_BALANCE_FN ?? "balance", + deductFunctionName: process.env.SOROBAN_BILLING_DEDUCT_FN ?? "deduct", }); return new BillingService(pool, sorobanClient); } function requireString(value: unknown, field: string): string { - if (typeof value !== 'string' || value.trim() === '') { + if (typeof value !== "string" || value.trim() === "") { throw new BadRequestError(`${field} is required`); } return value.trim(); } function requirePositiveAmount(value: unknown): string { - const amount = requireString(value, 'amountUsdc'); + const amount = requireString(value, "amountUsdc"); if (!/^\d+(\.\d{1,7})?$/.test(amount) || Number(amount) <= 0) { - throw new BadRequestError('amountUsdc must be a positive number with at most 7 decimal places'); + throw new BadRequestError( + "amountUsdc must be a positive number with at most 7 decimal places", + ); } return amount; } @@ -65,32 +83,32 @@ function requirePositiveAmount(value: unknown): string { function getPool(req: Request): Pool { const pool = req.app?.locals?.dbPool as Pool | undefined; if (!pool) { - throw new InternalServerError('Database pool is not configured'); + throw new InternalServerError("Database pool is not configured"); } return pool; } function sendSimulationFailure( res: Response, - result: Pick + result: Pick, ): void { - console.warn('Soroban simulation diagnostics:', result.simulationDetails); + console.warn("Soroban simulation diagnostics:", result.simulationDetails); res.status(502).json({ - error: 'Soroban simulation failed', - code: 'SIMULATION_FAILED', + error: "Soroban simulation failed", + code: "SIMULATION_FAILED", simulationDetails: redactSimulationDetails(result.simulationDetails), }); } router.post( - '/deduct', + "/deduct", requireAuth, idempotencyMiddleware, billingDeductHistogramMiddleware, async ( req: Request, res: Response, - next: NextFunction + next: NextFunction, ) => { try { const user = res.locals.authenticatedUser; @@ -100,15 +118,16 @@ router.post( } const body = req.body as BillingDeductBody; - const requestId = requireString(body.requestId, 'requestId'); - const apiId = requireString(body.apiId, 'apiId'); - const endpointId = requireString(body.endpointId, 'endpointId'); - const apiKeyId = requireString(body.apiKeyId, 'apiKeyId'); + const requestId = requireString(body.requestId, "requestId"); + const apiId = requireString(body.apiId, "apiId"); + const endpointId = requireString(body.endpointId, "endpointId"); + const apiKeyId = requireString(body.apiKeyId, "apiKeyId"); const amountUsdc = requirePositiveAmount(body.amountUsdc); const idempotencyKey = - typeof body.idempotencyKey === 'string' && body.idempotencyKey.trim() !== '' + typeof body.idempotencyKey === "string" && + body.idempotencyKey.trim() !== "" ? body.idempotencyKey.trim() - : req.get('Idempotency-Key') ?? undefined; + : (req.get("Idempotency-Key") ?? undefined); const billingService = createRouteBillingService(getPool(req)); const result = await billingService.deduct({ @@ -127,7 +146,12 @@ router.post( return; } - next(new PaymentRequiredError(result.error ?? 'Billing deduction failed', 'BILLING_DEDUCTION_FAILED')); + next( + new PaymentRequiredError( + result.error ?? "Billing deduction failed", + "BILLING_DEDUCTION_FAILED", + ), + ); return; } @@ -140,40 +164,45 @@ router.post( } catch (error) { if (error instanceof SorobanRpcError) { if (error.simulationDetails) { - console.warn('Soroban simulation diagnostics:', error.simulationDetails); + console.warn( + "Soroban simulation diagnostics:", + error.simulationDetails, + ); res.status(502).json({ - error: 'Soroban simulation failed', - code: 'SIMULATION_FAILED', + error: "Soroban simulation failed", + code: "SIMULATION_FAILED", simulationDetails: redactSimulationDetails(error.simulationDetails), }); return; } switch (error.category) { - case 'INSUFFICIENT_BALANCE': - next(new PaymentRequiredError(error.message, 'INSUFFICIENT_BALANCE')); + case "INSUFFICIENT_BALANCE": + next( + new PaymentRequiredError(error.message, "INSUFFICIENT_BALANCE"), + ); return; - case 'TIMEOUT': - next(new GatewayTimeoutError(error.message, 'SOROBAN_RPC_TIMEOUT')); + case "TIMEOUT": + next(new GatewayTimeoutError(error.message, "SOROBAN_RPC_TIMEOUT")); return; - case 'CONTRACT_ERROR': - case 'NETWORK_ERROR': - next(new BadGatewayError(error.message, 'SOROBAN_RPC_ERROR')); + case "CONTRACT_ERROR": + case "NETWORK_ERROR": + next(new BadGatewayError(error.message, "SOROBAN_RPC_ERROR")); return; } } next(error); } - } + }, ); router.get( - '/request/:requestId', + "/request/:requestId", requireAuth, async ( req: Request, res: Response, - next: NextFunction + next: NextFunction, ) => { try { const user = res.locals.authenticatedUser; @@ -182,12 +211,17 @@ router.get( return; } - const requestId = requireString(req.params.requestId, 'requestId'); + const requestId = requireString(req.params.requestId, "requestId"); const billingService = createRouteBillingService(getPool(req)); const result = await billingService.getByRequestId(requestId); if (!result) { - next(new NotFoundError('Billing request not found', 'BILLING_REQUEST_NOT_FOUND')); + next( + new NotFoundError( + "Billing request not found", + "BILLING_REQUEST_NOT_FOUND", + ), + ); return; } @@ -200,7 +234,7 @@ router.get( } catch (error) { next(error); } - } + }, ); export default router; diff --git a/src/routes/billing/deduct/bulk.test.ts b/src/routes/billing/deduct/bulk.test.ts new file mode 100644 index 0000000..cf8d4c9 --- /dev/null +++ b/src/routes/billing/deduct/bulk.test.ts @@ -0,0 +1,265 @@ +import express from 'express'; +import type { Application } from 'express'; +import request from 'supertest'; + +import { errorHandler } from '../../../middleware/errorHandler.js'; +import { requestIdMiddleware } from '../../../middleware/requestId.js'; +import { createBulkDeductRouter } from './bulk.js'; + +const mockBillingService = { + deductBulk: jest.fn(), +}; + +jest.mock('../../../logger.js', () => ({ + logger: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + audit: jest.fn(), + }, + getRequestId: jest.fn(), + runWithRequestContext: jest.fn((_, callback) => callback()), +})); + +describe('POST /api/billing/deduct/bulk', () => { + let app: Application; + + beforeEach(() => { + jest.clearAllMocks(); + + app = express(); + app.locals.dbPool = {}; + app.use(express.json()); + app.use(requestIdMiddleware); + app.use( + '/api/billing/deduct', + createBulkDeductRouter({ + createBillingService: () => mockBillingService, + }), + ); + app.use(errorHandler); + }); + + it('returns 401 without authentication', async () => { + const response = await request(app) + .post('/api/billing/deduct/bulk') + .send({ + entries: [ + { + requestId: 'req_1', + apiId: 'api_1', + endpointId: 'ep_1', + apiKeyId: 'key_1', + amountUsdc: '0.1', + }, + ], + }); + + expect(response.status).toBe(401); + expect(response.body).toMatchObject({ + code: 'UNAUTHORIZED', + requestId: expect.any(String), + }); + }); + + it('returns a standardized validation error when more than 100 entries are submitted', async () => { + const entries = Array.from({ length: 101 }, (_, index) => ({ + requestId: `req_${index}`, + apiId: 'api_1', + endpointId: 'ep_1', + apiKeyId: 'key_1', + amountUsdc: '0.01', + })); + + const response = await request(app) + .post('/api/billing/deduct/bulk') + .set('x-user-id', 'user_123') + .set('x-request-id', 'bulk-validation-request') + .send({ entries }); + + expect(response.status).toBe(400); + expect(response.body).toMatchObject({ + code: 'VALIDATION_ERROR', + message: 'Request validation failed', + requestId: 'bulk-validation-request', + }); + expect(response.body.details).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + field: 'body.entries', + }), + ]), + ); + expect(mockBillingService.deductBulk).not.toHaveBeenCalled(); + }); + + it('rejects duplicate requestId values within the same batch', async () => { + const response = await request(app) + .post('/api/billing/deduct/bulk') + .set('x-user-id', 'user_123') + .send({ + entries: [ + { + requestId: 'req_duplicate', + apiId: 'api_1', + endpointId: 'ep_1', + apiKeyId: 'key_1', + amountUsdc: '0.01', + }, + { + requestId: 'req_duplicate', + apiId: 'api_2', + endpointId: 'ep_2', + apiKeyId: 'key_2', + amountUsdc: '0.02', + }, + ], + }); + + expect(response.status).toBe(400); + expect(response.body.code).toBe('VALIDATION_ERROR'); + expect(response.body.details).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + field: 'body.entries[1].requestId', + message: 'requestId values must be unique within the batch', + }), + ]), + ); + }); + + it('returns 200 with per-entry results and forwards the authenticated user id', async () => { + mockBillingService.deductBulk.mockResolvedValue({ + success: true, + entryCount: 2, + deductedCount: 2, + totalDeductedAmountUsdc: '0.3', + stellarTxHash: 'tx_bulk_123', + results: [ + { + requestId: 'req_1', + usageEventId: '101', + stellarTxHash: 'tx_bulk_123', + alreadyProcessed: false, + deductionApplied: true, + reconciliationRequired: false, + }, + { + requestId: 'req_2', + usageEventId: '102', + stellarTxHash: 'tx_bulk_123', + alreadyProcessed: false, + deductionApplied: true, + reconciliationRequired: false, + }, + ], + }); + + const payload = { + idempotencyKey: 'bulk-key-1', + entries: [ + { + requestId: 'req_1', + apiId: 'api_1', + endpointId: 'ep_1', + apiKeyId: 'key_1', + amountUsdc: '0.1', + }, + { + requestId: 'req_2', + apiId: 'api_2', + endpointId: 'ep_2', + apiKeyId: 'key_2', + amountUsdc: '0.2', + }, + ], + }; + + const response = await request(app) + .post('/api/billing/deduct/bulk') + .set('x-user-id', 'user_123') + .send(payload); + + expect(response.status).toBe(200); + expect(response.body).toEqual({ + success: true, + entryCount: 2, + deductedCount: 2, + totalDeductedAmountUsdc: '0.3', + stellarTxHash: 'tx_bulk_123', + results: [ + { + requestId: 'req_1', + usageEventId: '101', + stellarTxHash: 'tx_bulk_123', + alreadyProcessed: false, + deductionApplied: true, + reconciliationRequired: false, + }, + { + requestId: 'req_2', + usageEventId: '102', + stellarTxHash: 'tx_bulk_123', + alreadyProcessed: false, + deductionApplied: true, + reconciliationRequired: false, + }, + ], + }); + + expect(mockBillingService.deductBulk).toHaveBeenCalledWith( + [ + { + requestId: 'req_1', + userId: 'user_123', + apiId: 'api_1', + endpointId: 'ep_1', + apiKeyId: 'key_1', + amountUsdc: '0.1', + }, + { + requestId: 'req_2', + userId: 'user_123', + apiId: 'api_2', + endpointId: 'ep_2', + apiKeyId: 'key_2', + amountUsdc: '0.2', + }, + ], + 'bulk-key-1', + ); + }); + + it('maps insufficient balance failures to the standard 402 envelope', async () => { + mockBillingService.deductBulk.mockResolvedValue({ + success: false, + entryCount: 1, + deductedCount: 0, + totalDeductedAmountUsdc: '0.5', + error: 'Insufficient balance: required 5000000 units, available 1 units', + results: [], + }); + + const response = await request(app) + .post('/api/billing/deduct/bulk') + .set('x-user-id', 'user_123') + .set('x-request-id', 'bulk-402-request') + .send({ + entries: [ + { + requestId: 'req_1', + apiId: 'api_1', + endpointId: 'ep_1', + apiKeyId: 'key_1', + amountUsdc: '0.5', + }, + ], + }); + + expect(response.status).toBe(402); + expect(response.body).toMatchObject({ + code: 'PAYMENT_REQUIRED', + requestId: 'bulk-402-request', + }); + }); +}); diff --git a/src/routes/billing/deduct/bulk.ts b/src/routes/billing/deduct/bulk.ts new file mode 100644 index 0000000..6bae8eb --- /dev/null +++ b/src/routes/billing/deduct/bulk.ts @@ -0,0 +1,242 @@ +import { Router } from "express"; +import type { NextFunction, Request, Response } from "express"; +import type { Pool } from "pg"; +import { z } from "zod"; + +import { + BadGatewayError, + BadRequestError, + GatewayTimeoutError, + InternalServerError, + PaymentRequiredError, + UnauthorizedError, +} from "../../../errors/index.js"; +import { logger } from "../../../logger.js"; +import { + requireAuth, + type AuthenticatedLocals, +} from "../../../middleware/requireAuth.js"; +import { idempotencyMiddleware } from "../../../middleware/idempotency.js"; +import { validate } from "../../../middleware/validate.js"; +import { + BillingService, + type BillingDeductRequest, +} from "../../../services/billing.js"; +import { + createSorobanRpcBillingClient, + SorobanRpcError, +} from "../../../services/sorobanBilling.js"; + +const MAX_BULK_DEDUCT_ENTRIES = 100; +const amountUsdcPattern = /^\d+(\.\d{1,7})?$/; + +const bulkDeductEntrySchema = z.object({ + requestId: z.string().trim().min(1, "requestId is required"), + apiId: z.string().trim().min(1, "apiId is required"), + endpointId: z.string().trim().min(1, "endpointId is required"), + apiKeyId: z.string().trim().min(1, "apiKeyId is required"), + amountUsdc: z + .string() + .trim() + .refine( + (value: string) => amountUsdcPattern.test(value) && Number(value) > 0, + "amountUsdc must be a positive number with at most 7 decimal places", + ), +}); + +const bulkDeductBodySchema = z + .object({ + entries: z + .array(bulkDeductEntrySchema) + .min(1, "At least one billing deduction entry is required") + .max( + MAX_BULK_DEDUCT_ENTRIES, + `Cannot deduct more than ${MAX_BULK_DEDUCT_ENTRIES} entries at once`, + ), + idempotencyKey: z.string().trim().min(1).max(255).optional(), + }) + .superRefine( + (body: { entries: Array<{ requestId: string }> }, ctx: z.RefinementCtx) => { + const seenRequestIds = new Set(); + + body.entries.forEach((entry: { requestId: string }, index: number) => { + if (seenRequestIds.has(entry.requestId)) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ["entries", index, "requestId"], + message: "requestId values must be unique within the batch", + }); + return; + } + + seenRequestIds.add(entry.requestId); + }); + }, + ); + +type BulkDeductBody = z.infer; + +type BulkBillingService = Pick; + +interface BulkDeductRouterOptions { + createBillingService?: (pool: Pool) => BulkBillingService; +} + +function getPool(req: Request): Pool { + const pool = req.app?.locals?.dbPool as Pool | undefined; + if (!pool) { + throw new InternalServerError("Database pool is not configured"); + } + return pool; +} + +function createRouteBillingService(pool: Pool): BillingService { + const sorobanClient = createSorobanRpcBillingClient({ + rpcUrl: + process.env.SOROBAN_BILLING_RPC_URL ?? + process.env.SOROBAN_RPC_URL ?? + "http://localhost:8000", + contractId: process.env.SOROBAN_BILLING_CONTRACT_ID ?? "vault_contract", + sourceAccount: process.env.SOROBAN_BILLING_SOURCE_ACCOUNT, + networkPassphrase: process.env.SOROBAN_BILLING_NETWORK_PASSPHRASE, + requestTimeoutMs: Number( + process.env.SOROBAN_BILLING_RPC_TIMEOUT_MS ?? 5_000, + ), + balanceFunctionName: process.env.SOROBAN_BILLING_BALANCE_FN ?? "balance", + deductFunctionName: process.env.SOROBAN_BILLING_DEDUCT_FN ?? "deduct", + }); + + return new BillingService(pool, sorobanClient); +} + +function mapBulkEntries( + userId: string, + body: BulkDeductBody, +): BillingDeductRequest[] { + return body.entries.map((entry: BulkDeductBody["entries"][number]) => ({ + requestId: entry.requestId, + userId, + apiId: entry.apiId, + endpointId: entry.endpointId, + apiKeyId: entry.apiKeyId, + amountUsdc: entry.amountUsdc, + })); +} + +function mapBulkFailure(error?: string, simulationDetails?: unknown): Error { + if (simulationDetails) { + return new BadGatewayError("Soroban simulation failed"); + } + + if (!error) { + return new PaymentRequiredError("Bulk billing deduction failed"); + } + + const lower = error.toLowerCase(); + + if (lower.includes("insufficient balance")) { + return new PaymentRequiredError(error); + } + + if (lower.includes("timeout")) { + return new GatewayTimeoutError(error); + } + + if (lower.includes("balance check failed")) { + return new BadGatewayError(error); + } + + if (lower.includes("single user")) { + return new BadRequestError(error); + } + + return new PaymentRequiredError(error); +} + +export function createBulkDeductRouter( + options: BulkDeductRouterOptions = {}, +): Router { + const router = Router(); + const createBillingService = + options.createBillingService ?? createRouteBillingService; + + router.post( + "/bulk", + requireAuth, + idempotencyMiddleware, + validate({ body: bulkDeductBodySchema }), + async ( + req: Request, + res: Response, + next: NextFunction, + ): Promise => { + try { + const user = res.locals.authenticatedUser; + if (!user) { + next(new UnauthorizedError()); + return; + } + + const body = bulkDeductBodySchema.parse(req.body); + const billingService = createBillingService(getPool(req)); + const result = await billingService.deductBulk( + mapBulkEntries(user.id, body), + body.idempotencyKey ?? req.get("Idempotency-Key") ?? undefined, + ); + + if (!result.success) { + logger.warn({ + event: "billing.bulk_deduct.failed", + correlationId: req.id ?? "unknown", + userId: user.id, + entryCount: body.entries.length, + error: result.error, + simulationDetails: result.simulationDetails, + }); + next(mapBulkFailure(result.error, result.simulationDetails)); + return; + } + + logger.info({ + event: "billing.bulk_deduct.succeeded", + correlationId: req.id ?? "unknown", + userId: user.id, + entryCount: result.entryCount, + deductedCount: result.deductedCount, + totalDeductedAmountUsdc: result.totalDeductedAmountUsdc, + stellarTxHash: result.stellarTxHash, + }); + + res.status(200).json({ + success: true, + entryCount: result.entryCount, + deductedCount: result.deductedCount, + totalDeductedAmountUsdc: result.totalDeductedAmountUsdc, + stellarTxHash: result.stellarTxHash, + results: result.results, + }); + } catch (error) { + if (error instanceof SorobanRpcError) { + switch (error.category) { + case "INSUFFICIENT_BALANCE": + next(new PaymentRequiredError(error.message)); + return; + case "TIMEOUT": + next(new GatewayTimeoutError(error.message)); + return; + case "CONTRACT_ERROR": + case "NETWORK_ERROR": + next(new BadGatewayError(error.message)); + return; + } + } + + next(error); + } + }, + ); + + return router; +} + +export default createBulkDeductRouter(); diff --git a/src/services/billing.bulk.test.ts b/src/services/billing.bulk.test.ts new file mode 100644 index 0000000..b178ab3 --- /dev/null +++ b/src/services/billing.bulk.test.ts @@ -0,0 +1,211 @@ +import assert from 'node:assert/strict'; +import type { Pool, PoolClient, QueryResult } from 'pg'; + +import { BillingService, billingInternals, type BillingDeductRequest, type SorobanClient } from './billing.js'; + +function makeQr(rows: Record[] = []): QueryResult { + return { rows, rowCount: rows.length, command: '', oid: 0, fields: [] } as QueryResult; +} + +function createMockSorobanClient(options?: { + balance?: string; + txHash?: string; + deductFailure?: Error; +}) { + let balanceCount = 0; + let deductCount = 0; + let lastDeductAmount: string | undefined; + + const client: SorobanClient = { + getBalance: async () => { + balanceCount += 1; + return { balance: options?.balance ?? '10000000' }; + }, + deductBalance: async (_userId, amount) => { + deductCount += 1; + lastDeductAmount = amount; + if (options?.deductFailure) { + throw options.deductFailure; + } + return { txHash: options?.txHash ?? 'tx_bulk_1' }; + }, + }; + + return { + client, + getBalanceCount: () => balanceCount, + getDeductCount: () => deductCount, + getLastDeductAmount: () => lastDeductAmount, + }; +} + +const baseRequests: BillingDeductRequest[] = [ + { + requestId: 'req_1', + userId: 'user_1', + apiId: 'api_1', + endpointId: 'ep_1', + apiKeyId: 'key_1', + amountUsdc: '0.1', + }, + { + requestId: 'req_2', + userId: 'user_1', + apiId: 'api_2', + endpointId: 'ep_2', + apiKeyId: 'key_2', + amountUsdc: '0.2', + }, +]; + +describe('BillingService.deductBulk', () => { + test('deducts only new entries and updates them with one tx hash', async () => { + const clientQueries: string[] = []; + const poolQueries: string[] = []; + + const client = { + query: async (sql: string, params: unknown[] = []) => { + clientQueries.push(sql); + if (sql === 'BEGIN' || sql === 'COMMIT' || sql === 'ROLLBACK') { + return makeQr(); + } + if (sql.includes('FOR UPDATE')) { + return makeQr([{ request_id: 'req_1', id: 55, stellar_tx_hash: 'tx_existing' }]); + } + if (sql.includes('INSERT INTO usage_events')) { + assert.equal(params[5], 'req_2'); + return makeQr([{ id: 99 }]); + } + throw new Error(`Unexpected client query: ${sql}`); + }, + release: () => {}, + } as unknown as PoolClient; + + const pool = { + connect: async () => client, + query: async (sql: string, params: unknown[] = []) => { + poolQueries.push(sql); + if (sql.includes('WHERE request_id = ANY')) { + return makeQr([{ request_id: 'req_1', id: 55, stellar_tx_hash: 'tx_existing' }]); + } + if (sql.includes('UPDATE usage_events')) { + assert.equal(params[0], 'tx_bulk_success'); + assert.deepEqual(params[1], ['99']); + return makeQr(); + } + throw new Error(`Unexpected pool query: ${sql}`); + }, + } as unknown as Pool; + + const soroban = createMockSorobanClient({ balance: '3000000', txHash: 'tx_bulk_success' }); + const svc = new BillingService(pool, soroban.client, { retryDelaysMs: [] }); + + const result = await svc.deductBulk(baseRequests, 'bulk-key-1'); + + assert.equal(result.success, true); + assert.equal(result.entryCount, 2); + assert.equal(result.deductedCount, 1); + assert.equal(result.totalDeductedAmountUsdc, '0.2'); + assert.equal(result.stellarTxHash, 'tx_bulk_success'); + assert.equal(soroban.getBalanceCount(), 1); + assert.equal(soroban.getDeductCount(), 1); + assert.equal(soroban.getLastDeductAmount(), '2000000'); + assert.equal(clientQueries.filter((sql) => sql.includes('INSERT INTO usage_events')).length, 1); + assert.equal(poolQueries.filter((sql) => sql.includes('UPDATE usage_events')).length, 1); + assert.deepEqual(result.results, [ + { + requestId: 'req_1', + usageEventId: '55', + stellarTxHash: 'tx_existing', + alreadyProcessed: true, + deductionApplied: true, + reconciliationRequired: false, + }, + { + requestId: 'req_2', + usageEventId: '99', + stellarTxHash: 'tx_bulk_success', + alreadyProcessed: false, + deductionApplied: true, + reconciliationRequired: false, + }, + ]); + }); + + test('returns existing rows without charging again when the full batch was already processed', async () => { + const client = { + query: async (sql: string) => { + if (sql === 'BEGIN' || sql === 'COMMIT' || sql === 'ROLLBACK') { + return makeQr(); + } + if (sql.includes('FOR UPDATE')) { + return makeQr([ + { request_id: 'req_1', id: 1, stellar_tx_hash: 'tx_1' }, + { request_id: 'req_2', id: 2, stellar_tx_hash: 'tx_2' }, + ]); + } + throw new Error(`Unexpected client query: ${sql}`); + }, + release: () => {}, + } as unknown as PoolClient; + + const pool = { + connect: async () => client, + query: async (sql: string) => { + if (sql.includes('WHERE request_id = ANY')) { + return makeQr([ + { request_id: 'req_1', id: 1, stellar_tx_hash: 'tx_1' }, + { request_id: 'req_2', id: 2, stellar_tx_hash: 'tx_2' }, + ]); + } + throw new Error(`Unexpected pool query: ${sql}`); + }, + } as unknown as Pool; + + const soroban = createMockSorobanClient({ balance: '99999999', txHash: 'unused' }); + const svc = new BillingService(pool, soroban.client, { retryDelaysMs: [] }); + + const result = await svc.deductBulk(baseRequests); + + assert.equal(result.success, true); + assert.equal(result.deductedCount, 0); + assert.equal(result.totalDeductedAmountUsdc, '0'); + assert.equal(soroban.getDeductCount(), 0); + assert.deepEqual(result.results.map((entry) => entry.alreadyProcessed), [true, true]); + }); + + test('fails before phase 1 when the aggregated new amount exceeds the available balance', async () => { + const pool = { + connect: async () => { + throw new Error('connect should not be called'); + }, + query: async (sql: string) => { + if (sql.includes('WHERE request_id = ANY')) { + return makeQr(); + } + throw new Error(`Unexpected pool query: ${sql}`); + }, + } as unknown as Pool; + + const soroban = createMockSorobanClient({ balance: '1' }); + const svc = new BillingService(pool, soroban.client, { retryDelaysMs: [] }); + + const result = await svc.deductBulk(baseRequests); + + assert.equal(result.success, false); + assert.equal(result.deductedCount, 0); + assert.equal(result.totalDeductedAmountUsdc, '0.3'); + assert.match(result.error ?? '', /Insufficient balance/); + assert.equal(soroban.getBalanceCount(), 1); + assert.equal(soroban.getDeductCount(), 0); + }); +}); + +describe('billingInternals.formatContractUnitsToUsdc', () => { + test('formats contract units without trailing zeroes', () => { + assert.equal(billingInternals.formatContractUnitsToUsdc(0n), '0'); + assert.equal(billingInternals.formatContractUnitsToUsdc(1n), '0.0000001'); + assert.equal(billingInternals.formatContractUnitsToUsdc(10000000n), '1'); + assert.equal(billingInternals.formatContractUnitsToUsdc(12340000n), '1.234'); + }); +}); diff --git a/src/services/billing.ts b/src/services/billing.ts index 577bf28..270ba83 100644 --- a/src/services/billing.ts +++ b/src/services/billing.ts @@ -29,9 +29,10 @@ * an operator reconciliation job can detect and either confirm or void. */ -import type { Pool, PoolClient } from 'pg'; -import type { SimulationDetails } from '../lib/simulationDiagnostics.js'; -import { DeveloperSemaphore } from '../utils/developerSemaphore.js'; +import { createHash } from "crypto"; +import type { Pool, PoolClient } from "pg"; +import type { SimulationDetails } from "../lib/simulationDiagnostics.js"; +import { DeveloperSemaphore } from "../utils/developerSemaphore.js"; const USDC_7_DECIMAL_FACTOR = 10_000_000n; const DEFAULT_RETRY_DELAYS_MS = [150, 500, 1_000]; @@ -69,6 +70,26 @@ export interface BillingDeductResult { simulationDetails?: SimulationDetails; } +export interface BillingBulkDeductEntryResult { + requestId: string; + usageEventId: string; + stellarTxHash?: string; + alreadyProcessed: boolean; + deductionApplied: boolean; + reconciliationRequired: boolean; +} + +export interface BillingBulkDeductResult { + success: boolean; + results: BillingBulkDeductEntryResult[]; + entryCount: number; + deductedCount: number; + totalDeductedAmountUsdc: string; + stellarTxHash?: string; + error?: string; + simulationDetails?: SimulationDetails; +} + export interface SorobanBalanceResult { balance: string; } @@ -97,16 +118,18 @@ export interface BillingServiceOptions { export function parseUsdcToContractUnits(amountUsdc: string): bigint { const trimmed = amountUsdc.trim(); if (!/^\d+(\.\d{1,7})?$/.test(trimmed)) { - throw new Error('amountUsdc must be a positive decimal with at most 7 fractional digits'); + throw new Error( + "amountUsdc must be a positive decimal with at most 7 fractional digits", + ); } - const [wholePart, fractionalPart = ''] = trimmed.split('.'); + const [wholePart, fractionalPart = ""] = trimmed.split("."); const whole = BigInt(wholePart); - const fraction = BigInt((fractionalPart + '0000000').slice(0, 7)); + const fraction = BigInt((fractionalPart + "0000000").slice(0, 7)); const result = whole * USDC_7_DECIMAL_FACTOR + fraction; if (result <= 0n) { - throw new Error('amountUsdc must be greater than zero'); + throw new Error("amountUsdc must be greater than zero"); } return result; @@ -115,31 +138,51 @@ export function parseUsdcToContractUnits(amountUsdc: string): bigint { export function isTransientSorobanError(error: unknown): boolean { const message = normalizeErrorMessage(error).toLowerCase(); return [ - 'timeout', - 'timed out', - 'socket hang up', - 'temporarily unavailable', - 'temporary outage', - 'econnreset', - 'econnrefused', - '503', - '429', - 'rate limit', - 'network error', - 'transport error', + "timeout", + "timed out", + "socket hang up", + "temporarily unavailable", + "temporary outage", + "econnreset", + "econnrefused", + "503", + "429", + "rate limit", + "network error", + "transport error", ].some((token) => message.includes(token)); } +export function formatContractUnitsToUsdc(amount: bigint): string { + const whole = amount / USDC_7_DECIMAL_FACTOR; + const fraction = amount % USDC_7_DECIMAL_FACTOR; + + if (fraction === 0n) { + return whole.toString(); + } + + return `${whole.toString()}.${fraction + .toString() + .padStart(7, "0") + .replace(/0+$/, "")}`; +} + +function buildBulkDeductIdempotencyKey(requestIds: string[]): string { + return createHash("sha256") + .update(requestIds.slice().sort().join(":")) + .digest("hex"); +} + function normalizeErrorMessage(error: unknown): string { if (error instanceof Error) return error.message; - if (typeof error === 'string') return error; - return 'Unknown error'; + if (typeof error === "string") return error; + return "Unknown error"; } function getSimulationDetails(error: unknown): SimulationDetails | undefined { - if (!error || typeof error !== 'object') return undefined; + if (!error || typeof error !== "object") return undefined; const details = (error as { simulationDetails?: unknown }).simulationDetails; - if (!details || typeof details !== 'object') return undefined; + if (!details || typeof details !== "object") return undefined; return details as SimulationDetails; } @@ -158,16 +201,35 @@ interface Phase1Result { stellarTxHash?: string; } +interface ExistingUsageEventRow { + request_id: string; + id: string; + stellar_tx_hash: string | null; +} + +interface BulkInsertedUsageEvent { + requestId: string; + usageEventId: string; +} + +interface BulkPhase1Result { + rowsByRequestId: Map; + inserted: BulkInsertedUsageEvent[]; +} + async function runPhase1( client: PoolClient, request: BillingDeductRequest, amountInContractUnits: bigint, ): Promise { - await client.query('BEGIN'); + await client.query("BEGIN"); // Lock the row (if it exists) so concurrent requests with the same // request_id serialise here rather than racing to INSERT. - const existingEvent = await client.query<{ id: string; stellar_tx_hash: string | null }>( + const existingEvent = await client.query<{ + id: string; + stellar_tx_hash: string | null; + }>( `SELECT id, stellar_tx_hash FROM usage_events WHERE request_id = $1 @@ -176,7 +238,7 @@ async function runPhase1( ); if (existingEvent.rows.length > 0) { - await client.query('COMMIT'); + await client.query("COMMIT"); return { alreadyExists: true, usageEventId: existingEvent.rows[0].id.toString(), @@ -213,7 +275,7 @@ async function runPhase1( ], ); - await client.query('COMMIT'); + await client.query("COMMIT"); return { alreadyExists: false, @@ -238,6 +300,107 @@ async function runPhase3( ); } +async function runPhase3Bulk( + pool: Pool, + usageEventIds: string[], + txHash: string, +): Promise { + if (usageEventIds.length === 0) { + return; + } + + await pool.query( + `UPDATE usage_events + SET stellar_tx_hash = $1 + WHERE id = ANY($2::bigint[])`, + [txHash, usageEventIds.map((id) => BigInt(id).toString())], + ); +} + +async function getUsageEventsByRequestIds( + pool: Pool, + requestIds: string[], +): Promise> { + if (requestIds.length === 0) { + return new Map(); + } + + const result = await pool.query( + `SELECT request_id, id, stellar_tx_hash + FROM usage_events + WHERE request_id = ANY($1::text[])`, + [requestIds], + ); + + return new Map( + result.rows.map((row: ExistingUsageEventRow) => [row.request_id, row]), + ); +} + +async function runPhase1Bulk( + client: PoolClient, + requests: BillingDeductRequest[], +): Promise { + await client.query("BEGIN"); + + const requestIds = requests.map((request) => request.requestId); + const existingEvents = await client.query( + `SELECT request_id, id, stellar_tx_hash + FROM usage_events + WHERE request_id = ANY($1::text[]) + FOR UPDATE`, + [requestIds], + ); + + const rowsByRequestId = new Map( + existingEvents.rows.map((row: ExistingUsageEventRow) => [ + row.request_id, + row, + ]), + ); + const inserted: BulkInsertedUsageEvent[] = []; + + for (const request of requests) { + if (rowsByRequestId.has(request.requestId)) { + continue; + } + + const insertResult = await client.query<{ id: string }>( + `INSERT INTO usage_events + (user_id, api_id, endpoint_id, api_key_id, amount_usdc, request_id, created_at) + VALUES ($1, $2, $3, $4, $5, $6, NOW()) + RETURNING id`, + [ + request.userId, + request.apiId, + request.endpointId, + request.apiKeyId, + request.amountUsdc, + request.requestId, + ], + ); + + const insertedRow: ExistingUsageEventRow = { + request_id: request.requestId, + id: insertResult.rows[0].id.toString(), + stellar_tx_hash: null, + }; + + rowsByRequestId.set(request.requestId, insertedRow); + inserted.push({ + requestId: request.requestId, + usageEventId: insertedRow.id, + }); + } + + await client.query("COMMIT"); + + return { + rowsByRequestId, + inserted, + }; +} + // --------------------------------------------------------------------------- // BillingService // --------------------------------------------------------------------------- @@ -261,7 +424,44 @@ export class BillingService { ); } - private async deductInternal(request: BillingDeductRequest): Promise { + async deductBulk( + requests: BillingDeductRequest[], + idempotencyKey?: string, + ): Promise { + if (requests.length === 0) { + return { + success: false, + results: [], + entryCount: 0, + deductedCount: 0, + totalDeductedAmountUsdc: "0", + error: "At least one billing deduction entry is required", + }; + } + + const [firstRequest] = requests; + const mixedUserIds = requests.some( + (request) => request.userId !== firstRequest.userId, + ); + if (mixedUserIds) { + return { + success: false, + results: [], + entryCount: requests.length, + deductedCount: 0, + totalDeductedAmountUsdc: "0", + error: "Bulk billing deductions must target a single user", + }; + } + + return billingConcurrencySemaphore.withSlot(firstRequest.userId, () => + this.deductBulkInternal(requests, idempotencyKey), + ); + } + + private async deductInternal( + request: BillingDeductRequest, + ): Promise { // --- Validate amount before touching the DB --- let amountInContractUnits: bigint; try { @@ -269,7 +469,7 @@ export class BillingService { } catch (error) { return { success: false, - usageEventId: '', + usageEventId: "", alreadyProcessed: false, deductionApplied: false, reconciliationRequired: false, @@ -297,7 +497,7 @@ export class BillingService { } catch (error) { return { success: false, - usageEventId: '', + usageEventId: "", alreadyProcessed: false, deductionApplied: false, reconciliationRequired: false, @@ -309,7 +509,7 @@ export class BillingService { if (availableBalance < amountInContractUnits) { return { success: false, - usageEventId: '', + usageEventId: "", alreadyProcessed: false, deductionApplied: false, reconciliationRequired: false, @@ -325,14 +525,21 @@ export class BillingService { } catch (error) { // Rollback on any Phase 1 failure (INSERT never committed) try { - await client.query('ROLLBACK'); + await client.query("ROLLBACK"); } catch { // ignore rollback errors } // Unique-constraint race: another concurrent request committed first - if (error instanceof Error && 'code' in error && (error as NodeJS.ErrnoException).code === '23505') { - const existing = await this.pool.query<{ id: string; stellar_tx_hash: string | null }>( + if ( + error instanceof Error && + "code" in error && + (error as { code?: string }).code === "23505" + ) { + const existing = await this.pool.query<{ + id: string; + stellar_tx_hash: string | null; + }>( `SELECT id, stellar_tx_hash FROM usage_events WHERE request_id = $1`, [request.requestId], ); @@ -350,7 +557,7 @@ export class BillingService { return { success: false, - usageEventId: '', + usageEventId: "", alreadyProcessed: false, deductionApplied: false, reconciliationRequired: false, @@ -424,8 +631,229 @@ export class BillingService { }; } + private async deductBulkInternal( + requests: BillingDeductRequest[], + idempotencyKey?: string, + ): Promise { + let totalRequestedAmount = 0n; + const amountByRequestId = new Map(); + + try { + for (const request of requests) { + const amount = parseUsdcToContractUnits(request.amountUsdc); + totalRequestedAmount += amount; + amountByRequestId.set(request.requestId, amount); + } + } catch (error) { + return { + success: false, + results: [], + entryCount: requests.length, + deductedCount: 0, + totalDeductedAmountUsdc: "0", + error: normalizeErrorMessage(error), + }; + } + + const requestIds = requests.map((request) => request.requestId); + const existingRows = await getUsageEventsByRequestIds( + this.pool, + requestIds, + ); + + let totalNewAmount = 0n; + for (const request of requests) { + if (!existingRows.has(request.requestId)) { + totalNewAmount += amountByRequestId.get(request.requestId) ?? 0n; + } + } + + if (totalNewAmount > 0n) { + let availableBalance: bigint; + try { + const balanceResult = await this.sorobanClient.getBalance( + requests[0].userId, + ); + availableBalance = BigInt(balanceResult.balance); + } catch (error) { + return { + success: false, + results: [], + entryCount: requests.length, + deductedCount: 0, + totalDeductedAmountUsdc: "0", + error: `Balance check failed: ${normalizeErrorMessage(error)}`, + simulationDetails: getSimulationDetails(error), + }; + } + + if (availableBalance < totalNewAmount) { + return { + success: false, + results: [], + entryCount: requests.length, + deductedCount: 0, + totalDeductedAmountUsdc: + formatContractUnitsToUsdc(totalRequestedAmount), + error: + `Insufficient balance: required ${totalNewAmount.toString()} units, ` + + `available ${availableBalance.toString()}`, + }; + } + } + + const client = await this.pool.connect(); + let phase1: BulkPhase1Result; + try { + phase1 = await runPhase1Bulk(client, requests); + } catch (error) { + try { + await client.query("ROLLBACK"); + } catch { + // ignore rollback errors + } + + if ( + error instanceof Error && + "code" in error && + (error as { code?: string }).code === "23505" + ) { + const recoveredRows = await getUsageEventsByRequestIds( + this.pool, + requestIds, + ); + return { + success: true, + results: requests.map((request) => { + const recovered = recoveredRows.get(request.requestId)!; + return { + requestId: request.requestId, + usageEventId: recovered.id.toString(), + stellarTxHash: recovered.stellar_tx_hash ?? undefined, + alreadyProcessed: true, + deductionApplied: Boolean(recovered.stellar_tx_hash), + reconciliationRequired: recovered.stellar_tx_hash === null, + }; + }), + entryCount: requests.length, + deductedCount: 0, + totalDeductedAmountUsdc: "0", + }; + } + + return { + success: false, + results: [], + entryCount: requests.length, + deductedCount: 0, + totalDeductedAmountUsdc: "0", + error: normalizeErrorMessage(error), + }; + } finally { + client.release(); + } + + const insertedRequestIds = new Set( + phase1.inserted.map((entry) => entry.requestId), + ); + let totalInsertedAmount = 0n; + for (const inserted of phase1.inserted) { + totalInsertedAmount += amountByRequestId.get(inserted.requestId) ?? 0n; + } + + if (phase1.inserted.length === 0) { + return { + success: true, + results: requests.map((request) => { + const existing = phase1.rowsByRequestId.get(request.requestId)!; + return { + requestId: request.requestId, + usageEventId: existing.id.toString(), + stellarTxHash: existing.stellar_tx_hash ?? undefined, + alreadyProcessed: true, + deductionApplied: Boolean(existing.stellar_tx_hash), + reconciliationRequired: existing.stellar_tx_hash === null, + }; + }), + entryCount: requests.length, + deductedCount: 0, + totalDeductedAmountUsdc: "0", + }; + } + + let deductResult: SorobanDeductResult; + try { + deductResult = await this.executeDeductWithRetry( + requests[0].userId, + totalInsertedAmount.toString(), + idempotencyKey ?? + buildBulkDeductIdempotencyKey(Array.from(insertedRequestIds)), + ); + } catch (error) { + return { + success: false, + results: requests.map((request) => { + const row = phase1.rowsByRequestId.get(request.requestId)!; + const wasInserted = insertedRequestIds.has(request.requestId); + return { + requestId: request.requestId, + usageEventId: row.id.toString(), + stellarTxHash: row.stellar_tx_hash ?? undefined, + alreadyProcessed: !wasInserted, + deductionApplied: !wasInserted && Boolean(row.stellar_tx_hash), + reconciliationRequired: wasInserted || row.stellar_tx_hash === null, + }; + }), + entryCount: requests.length, + deductedCount: 0, + totalDeductedAmountUsdc: "0", + error: normalizeErrorMessage(error), + simulationDetails: getSimulationDetails(error), + }; + } + + try { + await runPhase3Bulk( + this.pool, + phase1.inserted.map((entry) => entry.usageEventId), + deductResult.txHash, + ); + } catch (error) { + console.error( + `[BillingService] Bulk Phase 3 UPDATE failed for usageEventIds=` + + `${phase1.inserted.map((entry) => entry.usageEventId).join(",")} ` + + `txHash=${deductResult.txHash}: ${normalizeErrorMessage(error)}`, + ); + } + + return { + success: true, + results: requests.map((request) => { + const row = phase1.rowsByRequestId.get(request.requestId)!; + const wasInserted = insertedRequestIds.has(request.requestId); + return { + requestId: request.requestId, + usageEventId: row.id.toString(), + stellarTxHash: wasInserted + ? deductResult.txHash + : (row.stellar_tx_hash ?? undefined), + alreadyProcessed: !wasInserted, + deductionApplied: wasInserted || Boolean(row.stellar_tx_hash), + reconciliationRequired: !wasInserted && row.stellar_tx_hash === null, + }; + }), + entryCount: requests.length, + deductedCount: phase1.inserted.length, + totalDeductedAmountUsdc: formatContractUnitsToUsdc(totalInsertedAmount), + stellarTxHash: deductResult.txHash, + }; + } + async getByRequestId(requestId: string): Promise { - const result = await this.pool.query<{ id: string; stellar_tx_hash: string | null }>( + const result = await this.pool.query<{ + id: string; + stellar_tx_hash: string | null; + }>( `SELECT id, stellar_tx_hash FROM usage_events WHERE request_id = $1`, @@ -453,11 +881,18 @@ export class BillingService { for (let attempt = 0; attempt <= this.retryDelaysMs.length; attempt += 1) { try { - return await this.sorobanClient.deductBalance(userId, amount, idempotencyKey); + return await this.sorobanClient.deductBalance( + userId, + amount, + idempotencyKey, + ); } catch (error) { lastError = error; - if (!isTransientSorobanError(error) || attempt === this.retryDelaysMs.length) { + if ( + !isTransientSorobanError(error) || + attempt === this.retryDelaysMs.length + ) { break; } @@ -465,12 +900,15 @@ export class BillingService { } } - throw lastError instanceof Error ? lastError : new Error(normalizeErrorMessage(lastError)); + throw lastError instanceof Error + ? lastError + : new Error(normalizeErrorMessage(lastError)); } } // Exported for unit tests export const billingInternals = { parseUsdcToContractUnits, + formatContractUnitsToUsdc, isTransientSorobanError, };