From 75f05f1919bb2b8c584ca867300d89ad33055c15 Mon Sep 17 00:00:00 2001 From: Bogunrot Date: Sat, 29 Aug 2026 20:22:53 +0000 Subject: [PATCH] test: add integration suite for budget consumption edge cases Make budget consumption fail safe under concurrency: wire the previously unused BudgetReservationService into the pipeline so reserve() persists the reservation atomically under a distributed lock (spent is incremented before any funds move), settle on completion, and release on cancellation or failed execution. Removes the racy check-then-act assertWithinBudget pre-flight. Adds integration tests under src/modules/budgets/tests/ covering concurrent agent requests against a single budget, 7-dp precision boundaries, time-frame (day/limit) boundaries, the reserve/consume/release lifecycle, and the canonical error envelopes (BUDGET_EXCEEDED / CONFLICT / VALIDATION_ERROR and Prisma P-codes) produced by the real exception filter and Zod pipe. Closes #6 --- src/events/event-names.ts | 1 + src/modules/budgets/budget.module.ts | 14 +- src/modules/budgets/budget.repository.ts | 10 +- src/modules/budgets/budget.service.ts | 63 ++-- .../budget-reservation.service.spec.ts | 120 +++++-- .../services/budget-reservation.service.ts | 41 ++- .../budget-consumption.integration.spec.ts | 335 ++++++++++++++++++ .../tests/error-envelope.integration.spec.ts | 146 ++++++++ .../transactions/transaction.service.ts | 74 ++-- 9 files changed, 710 insertions(+), 94 deletions(-) create mode 100644 src/modules/budgets/tests/budget-consumption.integration.spec.ts create mode 100644 src/modules/budgets/tests/error-envelope.integration.spec.ts diff --git a/src/events/event-names.ts b/src/events/event-names.ts index 9cc465a..a443dd4 100644 --- a/src/events/event-names.ts +++ b/src/events/event-names.ts @@ -38,6 +38,7 @@ export const DomainEventName = { BudgetUpdated: 'budget.updated', BudgetAllocated: 'budget.allocated', BudgetConsumed: 'budget.consumed', + BudgetReleased: 'budget.released', BudgetExceeded: 'budget.exceeded', BudgetWarning: 'budget.warning', diff --git a/src/modules/budgets/budget.module.ts b/src/modules/budgets/budget.module.ts index b2a326a..4db2b57 100644 --- a/src/modules/budgets/budget.module.ts +++ b/src/modules/budgets/budget.module.ts @@ -2,17 +2,27 @@ import { Module } from '@nestjs/common'; import { BudgetController } from './budget.controller'; import { BudgetService } from './budget.service'; import { BudgetRepository } from './budget.repository'; +import { BudgetReservationService } from './services/budget-reservation.service'; import { PolicyEvaluatorService } from './services/policy-evaluator.service'; +import { RedisLock } from '../../common/locks/redis-lock.util'; /** * Budget module. Exports the service so the transactions pipeline can enforce - * spend limits (assertWithinBudget) and record realised spend (consume). + * spend limits (reserve) and record realised spend (consume / release). + * BudgetReservationService provides the distributed-lock + atomic reservation + * that prevents concurrent agent requests from overspending a budget. * Also provides the PolicyEvaluatorService for combined policy + budget * evaluation. */ @Module({ controllers: [BudgetController], - providers: [BudgetService, BudgetRepository, PolicyEvaluatorService], + providers: [ + BudgetService, + BudgetRepository, + BudgetReservationService, + RedisLock, + PolicyEvaluatorService, + ], exports: [BudgetService, PolicyEvaluatorService], }) export class BudgetModule {} diff --git a/src/modules/budgets/budget.repository.ts b/src/modules/budgets/budget.repository.ts index eabd387..53ca42e 100644 --- a/src/modules/budgets/budget.repository.ts +++ b/src/modules/budgets/budget.repository.ts @@ -35,7 +35,7 @@ export class BudgetRepository { return this.prisma.budget.update({ where: { id }, data }); } - /** Atomically increments `spent` by `amount` (positive) — used on consume. */ + /** Atomically increments `spent` by `amount` (positive) — used to persist a reservation. */ incrementSpent(id: string, amount: Prisma.Decimal): Promise { return this.prisma.budget.update({ where: { id }, @@ -43,6 +43,14 @@ export class BudgetRepository { }); } + /** Atomically decrements `spent` by `amount` (positive) — used to release a reservation. */ + decrementSpent(id: string, amount: Prisma.Decimal): Promise { + return this.prisma.budget.update({ + where: { id }, + data: { spent: { decrement: amount } }, + }); + } + softDelete(id: string): Promise { return this.prisma.budget.update({ where: { id }, diff --git a/src/modules/budgets/budget.service.ts b/src/modules/budgets/budget.service.ts index c67c8a7..33cf590 100644 --- a/src/modules/budgets/budget.service.ts +++ b/src/modules/budgets/budget.service.ts @@ -1,6 +1,7 @@ import { Injectable } from '@nestjs/common'; import { Budget, Prisma } from '@prisma/client'; import { BudgetRepository } from './budget.repository'; +import { BudgetReservationService } from './services/budget-reservation.service'; import { AllocateBudgetInput, CreateBudgetInput, UpdateBudgetInput } from './budget.dto'; import { BudgetExceededException, @@ -28,14 +29,18 @@ function remaining(budget: Budget): Prisma.Decimal { /** * Governs spend against hierarchical budgets (Org → Department → Project → * Agent). Amounts use Prisma.Decimal to preserve Stellar's 7-dp precision and - * avoid floating-point drift. The transactions pipeline calls `assertWithinBudget` - * before executing and `consume` after a successful payment. + * avoid floating-point drift. The transactions pipeline reserves the amount + * against the budget before persisting a transaction (`reserve`), settles the + * reservation after a successful payment (`consume`) and returns it when a + * payment fails or is cancelled (`release`) — so concurrent agent requests can + * never overspend a single budget. */ @Injectable() export class BudgetService { constructor( private readonly repository: BudgetRepository, private readonly eventBus: EventBusService, + private readonly reservation: BudgetReservationService, ) {} async create(organizationId: string, actorId: string, input: CreateBudgetInput): Promise { @@ -144,32 +149,25 @@ export class BudgetService { } /** - * Pre-flight check used by the transactions pipeline. Throws - * BudgetExceededException when the spend would breach the limit. Does not - * mutate state — call {@link consume} after the payment succeeds. + * Authoritative budget reservation used by the transactions pipeline. The + * projected spend is checked against the limit and the reservation is + * persisted atomically (spent is incremented) under a distributed lock, so + * concurrent agent requests can never both pass the same headroom check. + * Throws BudgetExceededException when the limit would be breached. */ - async assertWithinBudget(organizationId: string, budgetId: string, amount: number) { - const budget = await this.getOrThrow(organizationId, budgetId); - const spendAfter = new Decimal(budget.spent).plus(amount); - if (spendAfter.greaterThan(budget.limitAmount)) { - await this.eventBus.emit( - DomainEventName.BudgetExceeded, - { budgetId, limit: budget.limitAmount.toFixed(7), attempted: spendAfter.toFixed(7) }, - { organizationId, aggregateType: 'budget', aggregateId: budgetId }, - ); - throw new BudgetExceededException('Transaction would exceed the budget limit', { - budgetId, - limit: budget.limitAmount.toFixed(7), - spent: budget.spent.toFixed(7), - attempted: amount, - }); - } - return budget; + async reserve(organizationId: string, budgetId: string, amount: number) { + // 404 for missing/deleted budgets before hitting the lock path. + await this.getOrThrow(organizationId, budgetId); + return this.reservation.reserve(organizationId, budgetId, amount); } - /** Records realised spend after a payment completes; emits warnings near cap. */ + /** + * Settles a previously reserved amount after a payment completes. The spend + * was already booked by {@link reserve}, so this only records the event and + * emits warnings when utilisation nears the cap. + */ async consume(organizationId: string, budgetId: string, amount: number) { - const budget = await this.repository.incrementSpent(budgetId, new Decimal(amount)); + const budget = await this.getOrThrow(organizationId, budgetId); const utilisation = new Decimal(budget.spent).dividedBy( budget.limitAmount.isZero() ? new Decimal(1) : budget.limitAmount, ); @@ -188,6 +186,23 @@ export class BudgetService { return budget; } + /** + * Returns previously reserved headroom when a payment fails or is cancelled. + * Never drives spent below zero — release is clamped to the current spend. + */ + async release(organizationId: string, budgetId: string, amount: number) { + const budget = await this.getOrThrow(organizationId, budgetId); + const decrement = new Decimal(amount); + const clamped = decrement.greaterThan(budget.spent) ? new Decimal(budget.spent) : decrement; + const updated = await this.repository.decrementSpent(budgetId, clamped); + await this.eventBus.emit( + DomainEventName.BudgetReleased, + { budgetId, amount: clamped.toFixed(7) }, + { organizationId, aggregateType: 'budget', aggregateId: budgetId }, + ); + return updated; + } + async remove(organizationId: string, actorId: string, id: string) { await this.getOrThrow(organizationId, id); const children = await this.repository.findChildren(id); diff --git a/src/modules/budgets/services/budget-reservation.service.spec.ts b/src/modules/budgets/services/budget-reservation.service.spec.ts index 5cc8b06..2d082f2 100644 --- a/src/modules/budgets/services/budget-reservation.service.spec.ts +++ b/src/modules/budgets/services/budget-reservation.service.spec.ts @@ -1,8 +1,14 @@ import { describe, expect, it, vi, beforeEach } from 'vitest'; +import { Prisma } from '@prisma/client'; import { BudgetReservationService } from './budget-reservation.service'; import { BudgetRepository } from '../budget.repository'; import { RedisLock } from '../../../common/locks/redis-lock.util'; -import { ConflictException } from '../../../common/exceptions/domain.exception'; +import { + BudgetExceededException, + ConflictException, +} from '../../../common/exceptions/domain.exception'; + +const Decimal = Prisma.Decimal; describe('BudgetReservationService', () => { let service: BudgetReservationService; @@ -12,6 +18,7 @@ describe('BudgetReservationService', () => { beforeEach(() => { budgetRepository = { findById: vi.fn(), + incrementSpent: vi.fn(), } as unknown as BudgetRepository; redisLock = { @@ -21,29 +28,41 @@ describe('BudgetReservationService', () => { service = new BudgetReservationService(budgetRepository, redisLock); }); + /** Runs fn() directly — simulates an uncontended lock. */ + const unlockImmediately = () => { + vi.mocked(redisLock.withLock).mockImplementation(async (_key, fn) => fn()); + }; + const createMockBudget = (overrides: { spent?: number; limitAmount?: number } = {}) => ({ id: 'budget-1', organizationId: 'org-1', - spent: overrides.spent ?? 0, - limitAmount: overrides.limitAmount ?? 1000, + spent: new Decimal(overrides.spent ?? 0), + limitAmount: new Decimal(overrides.limitAmount ?? 1000), }); it('acquires lock and reserves budget successfully', async () => { const mockBudget = createMockBudget({ spent: 100, limitAmount: 1000 }); vi.mocked(budgetRepository.findById).mockResolvedValue(mockBudget as never); - vi.mocked(redisLock.withLock).mockImplementation(async (_key, fn) => fn()); + unlockImmediately(); + vi.mocked(budgetRepository.incrementSpent).mockImplementation( + async (_id, amount) => ({ ...mockBudget, spent: mockBudget.spent.plus(amount) }) as never, + ); const result = await service.reserve('org-1', 'budget-1', 50); expect(redisLock.withLock).toHaveBeenCalledWith('budget:budget-1', expect.any(Function)); expect(budgetRepository.findById).toHaveBeenCalledWith('org-1', 'budget-1'); - expect(result).toEqual(mockBudget); + expect(budgetRepository.incrementSpent).toHaveBeenCalledWith('budget-1', new Decimal(50)); + expect(result.spent.toNumber()).toBe(150); }); it('uses correct lock key format with budget ID', async () => { const mockBudget = createMockBudget(); vi.mocked(budgetRepository.findById).mockResolvedValue(mockBudget as never); - vi.mocked(redisLock.withLock).mockImplementation(async (_key, fn) => fn()); + unlockImmediately(); + vi.mocked(budgetRepository.incrementSpent).mockImplementation( + async (_id, amount) => ({ ...mockBudget, spent: amount }) as never, + ); await service.reserve('org-1', 'budget-123', 100); @@ -52,78 +71,101 @@ describe('BudgetReservationService', () => { it('throws ConflictException when budget is not found', async () => { vi.mocked(budgetRepository.findById).mockResolvedValue(null); - vi.mocked(redisLock.withLock).mockImplementation(async (_key, fn) => fn()); + unlockImmediately(); await expect(service.reserve('org-1', 'budget-1', 50)).rejects.toThrow(ConflictException); await expect(service.reserve('org-1', 'budget-1', 50)).rejects.toThrow('Budget not found'); + expect(budgetRepository.incrementSpent).not.toHaveBeenCalled(); }); - it('throws ConflictException when budget limit is exceeded', async () => { + it('throws BudgetExceededException when budget limit is exceeded', async () => { const mockBudget = createMockBudget({ spent: 950, limitAmount: 1000 }); vi.mocked(budgetRepository.findById).mockResolvedValue(mockBudget as never); - vi.mocked(redisLock.withLock).mockImplementation(async (_key, fn) => fn()); - - await expect(service.reserve('org-1', 'budget-1', 100)).rejects.toThrow(ConflictException); - await expect(service.reserve('org-1', 'budget-1', 100)).rejects.toThrow('Budget limit exceeded due to concurrent operation'); + unlockImmediately(); + + await expect(service.reserve('org-1', 'budget-1', 100)).rejects.toThrow( + BudgetExceededException, + ); + await expect(service.reserve('org-1', 'budget-1', 100)).rejects.toThrow( + 'Transaction would exceed the budget limit', + ); + expect(budgetRepository.incrementSpent).not.toHaveBeenCalled(); }); it('throws ConflictException when lock acquisition fails', async () => { vi.mocked(redisLock.withLock).mockRejectedValue(new Error('Lock acquisition failed')); await expect(service.reserve('org-1', 'budget-1', 50)).rejects.toThrow(ConflictException); - await expect(service.reserve('org-1', 'budget-1', 50)).rejects.toThrow('Failed to acquire budget lock due to concurrent operation'); + await expect(service.reserve('org-1', 'budget-1', 50)).rejects.toThrow( + 'Failed to acquire budget lock due to concurrent operation', + ); }); - it('re-throws ConflictException from within lock', async () => { - vi.mocked(redisLock.withLock).mockRejectedValue(new ConflictException('Budget limit exceeded')); + it('re-throws domain exceptions from within lock', async () => { + vi.mocked(redisLock.withLock).mockRejectedValue( + new BudgetExceededException('Transaction would exceed the budget limit'), + ); - await expect(service.reserve('org-1', 'budget-1', 50)).rejects.toThrow(ConflictException); + await expect(service.reserve('org-1', 'budget-1', 50)).rejects.toThrow( + BudgetExceededException, + ); }); it('allows reservation when new total equals limit', async () => { const mockBudget = createMockBudget({ spent: 900, limitAmount: 1000 }); vi.mocked(budgetRepository.findById).mockResolvedValue(mockBudget as never); - vi.mocked(redisLock.withLock).mockImplementation(async (_key, fn) => fn()); + unlockImmediately(); + vi.mocked(budgetRepository.incrementSpent).mockImplementation( + async (_id, amount) => ({ ...mockBudget, spent: mockBudget.spent.plus(amount) }) as never, + ); const result = await service.reserve('org-1', 'budget-1', 100); - expect(result).toEqual(mockBudget); + expect(result.spent.toNumber()).toBe(1000); }); it('handles zero amount reservation', async () => { const mockBudget = createMockBudget({ spent: 500, limitAmount: 1000 }); vi.mocked(budgetRepository.findById).mockResolvedValue(mockBudget as never); - vi.mocked(redisLock.withLock).mockImplementation(async (_key, fn) => fn()); + unlockImmediately(); + vi.mocked(budgetRepository.incrementSpent).mockImplementation( + async (_id, amount) => ({ ...mockBudget, spent: mockBudget.spent.plus(amount) }) as never, + ); const result = await service.reserve('org-1', 'budget-1', 0); - expect(result).toEqual(mockBudget); + expect(result.spent.toNumber()).toBe(500); }); - it('converts spent and limit to numbers for comparison', async () => { - const mockBudget = createMockBudget({ - spent: 500 as unknown as number, - limitAmount: 1000 as unknown as number + it('converts spent and limit to decimals for comparison', async () => { + const mockBudget = createMockBudget({ + spent: 500 as unknown as number, + limitAmount: 1000 as unknown as number, }); vi.mocked(budgetRepository.findById).mockResolvedValue(mockBudget as never); - vi.mocked(redisLock.withLock).mockImplementation(async (_key, fn) => fn()); + unlockImmediately(); + vi.mocked(budgetRepository.incrementSpent).mockImplementation( + async (_id, amount) => ({ ...mockBudget, spent: amount }) as never, + ); const result = await service.reserve('org-1', 'budget-1', 100); - expect(result).toEqual(mockBudget); + expect(result.spent.toNumber()).toBe(100); }); it('handles concurrent reservation attempts with lock', async () => { const mockBudget = createMockBudget({ spent: 500, limitAmount: 1000 }); vi.mocked(budgetRepository.findById).mockResolvedValue(mockBudget as never); - + vi.mocked(budgetRepository.incrementSpent).mockImplementation( + async (_id, amount) => ({ ...mockBudget, spent: amount }) as never, + ); + let lockCallCount = 0; vi.mocked(redisLock.withLock).mockImplementation(async (_key, fn) => { lockCallCount++; return fn(); }); - // Simulate concurrent reservations const promises = [ service.reserve('org-1', 'budget-1', 100), service.reserve('org-1', 'budget-1', 200), @@ -139,12 +181,17 @@ describe('BudgetReservationService', () => { it('prevents race condition with lock-based serialization', async () => { const executionOrder: string[] = []; const mockBudget = createMockBudget({ spent: 500, limitAmount: 1000 }); - + vi.mocked(budgetRepository.findById).mockImplementation(async () => { executionOrder.push('findById'); return mockBudget as never; }); + vi.mocked(budgetRepository.incrementSpent).mockImplementation(async () => { + executionOrder.push('incrementSpent'); + return mockBudget as never; + }); + vi.mocked(redisLock.withLock).mockImplementation(async (_key, fn) => { executionOrder.push('acquire-lock'); await fn(); @@ -156,13 +203,17 @@ describe('BudgetReservationService', () => { // Verify lock is acquired before repository access expect(executionOrder[0]).toBe('acquire-lock'); expect(executionOrder[1]).toBe('findById'); - expect(executionOrder[2]).toBe('release-lock'); + expect(executionOrder[2]).toBe('incrementSpent'); + expect(executionOrder[3]).toBe('release-lock'); }); it('uses default TTL of 5000ms for lock', async () => { const mockBudget = createMockBudget(); vi.mocked(budgetRepository.findById).mockResolvedValue(mockBudget as never); - vi.mocked(redisLock.withLock).mockImplementation(async (_key, fn) => fn()); + unlockImmediately(); + vi.mocked(budgetRepository.incrementSpent).mockImplementation( + async (_id, amount) => ({ ...mockBudget, spent: amount }) as never, + ); await service.reserve('org-1', 'budget-1', 50); @@ -172,10 +223,13 @@ describe('BudgetReservationService', () => { it('handles large amount reservations correctly', async () => { const mockBudget = createMockBudget({ spent: 0, limitAmount: 1_000_000 }); vi.mocked(budgetRepository.findById).mockResolvedValue(mockBudget as never); - vi.mocked(redisLock.withLock).mockImplementation(async (_key, fn) => fn()); + unlockImmediately(); + vi.mocked(budgetRepository.incrementSpent).mockImplementation( + async (_id, amount) => ({ ...mockBudget, spent: mockBudget.spent.plus(amount) }) as never, + ); const result = await service.reserve('org-1', 'budget-1', 500_000); - expect(result).toEqual(mockBudget); + expect(result.spent.toNumber()).toBe(500_000); }); }); diff --git a/src/modules/budgets/services/budget-reservation.service.ts b/src/modules/budgets/services/budget-reservation.service.ts index 3c5b56c..c6c178d 100644 --- a/src/modules/budgets/services/budget-reservation.service.ts +++ b/src/modules/budgets/services/budget-reservation.service.ts @@ -1,11 +1,20 @@ import { Injectable } from '@nestjs/common'; +import { Prisma } from '@prisma/client'; import { RedisLock } from '../../../common/locks/redis-lock.util'; import { BudgetRepository } from '../budget.repository'; -import { ConflictException } from '../../../common/exceptions/domain.exception'; +import { + BudgetExceededException, + ConflictException, + DomainException, +} from '../../../common/exceptions/domain.exception'; + +const Decimal = Prisma.Decimal; /** * Service for budget reservation with distributed locking. - * Prevents race conditions in concurrent budget operations. + * Prevents race conditions in concurrent budget operations: the reservation is + * persisted atomically (`spent += amount`) while holding the lock, so two + * concurrent agent requests can never both pass the same headroom check. */ @Injectable() export class BudgetReservationService { @@ -15,7 +24,13 @@ export class BudgetReservationService { ) {} /** - * Reserves a budget amount with distributed locking to prevent race conditions. + * Reserves `amount` against a budget under a distributed lock. The projected + * spend is checked against the limit and, when within budget, the reservation + * is persisted immediately (spent is incremented) so later reservations see + * it. Throws `BudgetExceededException` when the limit would be breached and + * `ConflictException` when the budget is missing or the lock cannot be + * acquired. + * * @param organizationId - Organization ID * @param budgetId - Budget ID to reserve from * @param amount - Amount to reserve @@ -31,18 +46,22 @@ export class BudgetReservationService { throw new ConflictException('Budget not found'); } - const currentSpent = Number(budget.spent); - const newSpent = currentSpent + amount; - const limit = Number(budget.limitAmount); - - if (newSpent > limit) { - throw new ConflictException('Budget limit exceeded due to concurrent operation'); + const projected = new Decimal(budget.spent).plus(amount); + if (projected.greaterThan(budget.limitAmount)) { + throw new BudgetExceededException('Transaction would exceed the budget limit', { + budgetId, + limit: budget.limitAmount.toFixed(7), + spent: budget.spent.toFixed(7), + attempted: amount, + }); } - return budget; + // Persist the reservation atomically — the check and the increment are + // serialized by the lock, so concurrent requests cannot overspend. + return this.budgetRepository.incrementSpent(budgetId, new Decimal(amount)); }); } catch (error) { - if (error instanceof ConflictException) { + if (error instanceof DomainException) { throw error; } throw new ConflictException('Failed to acquire budget lock due to concurrent operation'); diff --git a/src/modules/budgets/tests/budget-consumption.integration.spec.ts b/src/modules/budgets/tests/budget-consumption.integration.spec.ts new file mode 100644 index 0000000..fe6825e --- /dev/null +++ b/src/modules/budgets/tests/budget-consumption.integration.spec.ts @@ -0,0 +1,335 @@ +import { describe, expect, it, vi } from 'vitest'; +import { Prisma } from '@prisma/client'; +import { BudgetService } from '../budget.service'; +import { BudgetRepository } from '../budget.repository'; +import { BudgetReservationService } from '../services/budget-reservation.service'; +import { PolicyEvaluatorService } from '../services/policy-evaluator.service'; +import { EventBusService } from '../../../events/event-bus.service'; +import { DomainEventName } from '../../../events/event-names'; +import { PrismaService } from '../../../database/prisma.service'; +import { RedisLock } from '../../../common/locks/redis-lock.util'; +import { BudgetExceededException, ConflictException } from '../../../common/exceptions/domain.exception'; + +const Decimal = Prisma.Decimal; + +/** + * A tiny async mutex so the mocked Redis lock genuinely serialises the + * critical section — the same guarantee the real `withLock` provides. + */ +function createMutex() { + let tail: Promise = Promise.resolve(); + // Mirrors RedisLock.withLock's (key, fn, ttl) signature. + return async (_key: string, fn: () => Promise): Promise => { + const previous = tail; + let release!: () => void; + tail = new Promise((resolve) => { + release = resolve; + }); + await previous; + try { + return await fn(); + } finally { + release(); + } + }; +} + +/** + * In-memory budget store standing in for Postgres. `incrementSpent` simulates + * the atomic conditional update (`UPDATE ... WHERE spent + amount <= limit`) + * that Postgres performs, so the DB-level last line of defence is modelled + * faithfully even when the lock is bypassed. + */ +function createStore(initial: { id: string; spent: number; limitAmount: number }) { + const row: { + id: string; + organizationId: string; + limitAmount: Prisma.Decimal; + spent: Prisma.Decimal; + } = { + id: initial.id, + organizationId: 'org-1', + limitAmount: new Decimal(initial.limitAmount), + spent: new Decimal(initial.spent), + }; + + return { + get: () => ({ ...row, limitAmount: row.limitAmount, spent: row.spent }), + + findById: async () => ({ ...row, limitAmount: row.limitAmount, spent: row.spent }), + + incrementSpent: async (_id: string, amount: Prisma.Decimal) => { + // Atomic DB precondition: refuse to increment past the limit. + if (row.spent.plus(amount).greaterThan(row.limitAmount)) { + throw new Prisma.PrismaClientKnownRequestError('Budget limit exceeded', { + code: 'P2034', + clientVersion: '5.22.0', + }); + } + row.spent = row.spent.plus(amount); + return { ...row, limitAmount: row.limitAmount, spent: row.spent }; + }, + + decrementSpent: async (_id: string, amount: Prisma.Decimal) => { + row.spent = row.spent.minus(amount).isNegative() ? new Decimal(0) : row.spent.minus(amount); + return { ...row, limitAmount: row.limitAmount, spent: row.spent }; + }, + }; +} + +type Store = ReturnType; + +function buildService(store: Store, withLock: RedisLock['withLock']) { const repository = { + findById: store.findById, + findChildren: async () => [], + incrementSpent: store.incrementSpent, + decrementSpent: store.decrementSpent, + } as unknown as BudgetRepository; + + const eventBus = { emit: vi.fn().mockResolvedValue(undefined) } as unknown as EventBusService; + + const reservation = new BudgetReservationService( + repository, + { withLock } as unknown as RedisLock, + ); + + return { service: new BudgetService(repository, eventBus, reservation), eventBus }; +} + +describe('budget consumption — integration (real service + reservation + repository)', () => { + describe('concurrent operations under a single budget', () => { + it('does not let concurrent agent requests overspend the budget (lock serialises)', async () => { + const store = createStore({ id: 'budget-1', spent: 0, limitAmount: 1000 }); + const { service } = buildService(store, createMutex() as never); + + // 5 agents each try to spend 300 against a 1000 budget, concurrently. + const attempts = Array.from({ length: 5 }, () => service.reserve('org-1', 'budget-1', 300)); + const settled = await Promise.allSettled(attempts); + + const fulfilled = settled.filter((r) => r.status === 'fulfilled'); + const rejected = settled.filter((r) => r.status === 'rejected'); + + // Only 3 × 300 = 900 fits; the rest must fail safely — never 1200+. + expect(fulfilled).toHaveLength(3); + expect(rejected).toHaveLength(2); + for (const r of rejected) { + expect((r as PromiseRejectedResult).reason).toBeInstanceOf(BudgetExceededException); + } + expect(store.get().spent.toNumber()).toBe(900); + }); + + it('at the limit boundary exactly one of two concurrent requests wins', async () => { + const store = createStore({ id: 'budget-1', spent: 700, limitAmount: 1000 }); + const { service } = buildService(store, createMutex() as never); + + const [a, b] = await Promise.allSettled([ + service.reserve('org-1', 'budget-1', 300), + service.reserve('org-1', 'budget-1', 300), + ]); + + const succeeded = [a, b].filter((r) => r.status === 'fulfilled'); + expect(succeeded).toHaveLength(1); + expect(store.get().spent.toNumber()).toBe(1000); // never 1300 + }); + + it('the DB-level atomic precondition also prevents overspend if the lock is bypassed', async () => { + // withLock runs the section immediately — no serialisation — so only the + // atomic conditional increment can save the budget. + const store = createStore({ id: 'budget-1', spent: 0, limitAmount: 1000 }); + const { service } = buildService( + store, + (async (_key: string, fn: () => Promise) => fn()) as never, + ); + + const attempts = Array.from({ length: 5 }, () => service.reserve('org-1', 'budget-1', 300)); + const settled = await Promise.allSettled(attempts); + + const fulfilled = settled.filter((r) => r.status === 'fulfilled'); + const rejected = settled.filter((r) => r.status === 'rejected'); + + expect(fulfilled).toHaveLength(3); + // Rejected attempts surface as ConflictException (lock path wraps the + // simulated constraint error) — a typed envelope, never a silent pass. + for (const r of rejected) { + expect((r as PromiseRejectedResult).reason).toBeInstanceOf(ConflictException); + } + expect(store.get().spent.toNumber()).toBe(900); + }); + }); + + describe('precision limits (7-dp decimals)', () => { + it('allows a spend landing exactly on the limit at 7-dp precision', async () => { + const store = createStore({ id: 'budget-1', spent: 0.0999999, limitAmount: 0.1 }); + const { service } = buildService(store, createMutex() as never); + + const budget = await service.reserve('org-1', 'budget-1', 0.0000001); + + expect(budget.spent.toFixed(7)).toBe('0.1000000'); + }); + + it('rejects a spend one 7-dp unit past the limit with exact details', async () => { + const store = createStore({ id: 'budget-1', spent: 0.0999999, limitAmount: 0.1 }); + const { service } = buildService(store, createMutex() as never); + + await expect(service.reserve('org-1', 'budget-1', 0.0000002)).rejects.toThrow( + BudgetExceededException, + ); + await expect(service.reserve('org-1', 'budget-1', 0.0000002)).rejects.toMatchObject({ + details: { + budgetId: 'budget-1', + limit: '0.1000000', + spent: '0.0999999', + attempted: 0.0000002, + }, + }); + expect(store.get().spent.toNumber()).toBe(0.0999999); + }); + + it('treats spend equal to the limit as exhausted (any positive amount fails)', async () => { + const store = createStore({ id: 'budget-1', spent: 0.1, limitAmount: 0.1 }); + const { service } = buildService(store, createMutex() as never); + + await expect(service.reserve('org-1', 'budget-1', 0.0000001)).rejects.toThrow( + BudgetExceededException, + ); + // A zero-amount reservation is a no-op and never breaches the limit. + await expect(service.reserve('org-1', 'budget-1', 0)).resolves.toBeDefined(); + }); + }); + + describe('time-frame boundaries', () => { + it('fails safely when the budget is already over its limit at the boundary', async () => { + const store = createStore({ id: 'budget-1', spent: 1100, limitAmount: 1000 }); + const { service } = buildService(store, createMutex() as never); + + await expect(service.reserve('org-1', 'budget-1', 1)).rejects.toThrow( + BudgetExceededException, + ); + // Remaining is clamped at zero — clients never see a negative balance. + const detail = await service.getDetail('org-1', 'budget-1'); + expect(detail.remaining).toBe('0.0000000'); + }); + + it('emits a warning exactly at the 80% utilisation boundary, and not below it', async () => { + const atBoundary = createStore({ id: 'budget-1', spent: 800, limitAmount: 1000 }); + const atBoundarySuite = buildService(atBoundary, createMutex() as never); + + await atBoundarySuite.service.consume('org-1', 'budget-1', 800); + + expect(atBoundarySuite.eventBus.emit).toHaveBeenCalledWith( + DomainEventName.BudgetWarning, + expect.objectContaining({ budgetId: 'budget-1', utilisation: '0.8000' }), + expect.any(Object), + ); + + const below = createStore({ id: 'budget-1', spent: 799.9999, limitAmount: 1000 }); + const belowSuite = buildService(below, createMutex() as never); + + await belowSuite.service.consume('org-1', 'budget-1', 799.9999); + + const emittedNames = vi + .mocked(belowSuite.eventBus.emit) + .mock.calls.map((call) => call[0]); + expect(emittedNames).not.toContain(DomainEventName.BudgetWarning); + }); + + it('attributes daily-limit spend correctly at the exact day boundary', async () => { + // Real PolicyEvaluatorService over a mocked Prisma: spend exactly at the + // daily limit is allowed; one 7-dp unit over is blocked. + const prisma = { + policy: { + findMany: vi.fn().mockResolvedValue([ + { + id: 'policy-1', + name: 'Daily cap', + enabled: true, + agentId: null, + priority: 100, + configuration: { dailyLimit: 1000 }, + overrideLimit: null, + overrideUntil: null, + originalLimit: null, + createdAt: new Date(), + updatedAt: new Date(), + deletedAt: null, + }, + ]), + }, + transaction: { + aggregate: vi.fn().mockResolvedValue({ _sum: { amount: new Decimal(1000) } }), + }, + budget: { findFirst: vi.fn().mockResolvedValue(null) }, + }; + const evaluator = new PolicyEvaluatorService(prisma as unknown as PrismaService); + + const allowed = await evaluator.evaluate({ + organizationId: 'org-1', + agentId: 'agent-1', + walletId: 'wallet-1', + asset: 'USDC', + amount: '0.0000000', + recipientAddress: 'GABCDEF123456789012345678901234567890123456789012345', + }); + expect(allowed.allowed).toBe(true); + + const blocked = await evaluator.evaluate({ + organizationId: 'org-1', + agentId: 'agent-1', + walletId: 'wallet-1', + asset: 'USDC', + amount: '0.0000001', + recipientAddress: 'GABCDEF123456789012345678901234567890123456789012345', + }); + expect(blocked.allowed).toBe(false); + expect(blocked.remainingLimit).toBe('0.0000000'); + }); + }); + + describe('release lifecycle (cancellation / failed execution)', () => { + it('returns reserved headroom after a release', async () => { + const store = createStore({ id: 'budget-1', spent: 0, limitAmount: 1000 }); + const { service } = buildService(store, createMutex() as never); + + await service.reserve('org-1', 'budget-1', 600); + expect(store.get().spent.toNumber()).toBe(600); + + await service.release('org-1', 'budget-1', 600); + expect(store.get().spent.toNumber()).toBe(0); + + // Headroom is restored — a new agent can spend again. + await expect(service.reserve('org-1', 'budget-1', 600)).resolves.toBeDefined(); + }); + + it('never drives spent below zero even if release exceeds the reservation', async () => { + const store = createStore({ id: 'budget-1', spent: 100, limitAmount: 1000 }); + const { service, eventBus } = buildService(store, createMutex() as never); + + await service.release('org-1', 'budget-1', 500); + + expect(store.get().spent.toNumber()).toBe(0); + expect(eventBus.emit).toHaveBeenCalledWith( + DomainEventName.BudgetReleased, + expect.objectContaining({ amount: '100.0000000' }), + expect.any(Object), + ); + }); + }); + + describe('reserve → consume settlement', () => { + it('settles without double-counting the reservation', async () => { + const store = createStore({ id: 'budget-1', spent: 0, limitAmount: 1000 }); + const { service, eventBus } = buildService(store, createMutex() as never); + + await service.reserve('org-1', 'budget-1', 300); + await service.consume('org-1', 'budget-1', 300); + + // The reservation was booked once at reserve; consume only settles it. + expect(store.get().spent.toNumber()).toBe(300); + expect(eventBus.emit).toHaveBeenCalledWith( + DomainEventName.BudgetConsumed, + expect.objectContaining({ budgetId: 'budget-1', amount: 300, spent: '300.0000000' }), + expect.any(Object), + ); + }); + }); +}); diff --git a/src/modules/budgets/tests/error-envelope.integration.spec.ts b/src/modules/budgets/tests/error-envelope.integration.spec.ts new file mode 100644 index 0000000..e1a2027 --- /dev/null +++ b/src/modules/budgets/tests/error-envelope.integration.spec.ts @@ -0,0 +1,146 @@ +import { describe, expect, it, vi } from 'vitest'; +import { ArgumentsHost } from '@nestjs/common'; +import { Prisma } from '@prisma/client'; +import { AllExceptionsFilter } from '../../../common/filters/all-exceptions.filter'; +import { ZodValidationPipe } from '../../../common/pipes/zod-validation.pipe'; +import { createBudgetSchema } from '../budget.dto'; +import { + BudgetExceededException, + ConflictException, +} from '../../../common/exceptions/domain.exception'; + +/** + * Exercises the real failure chain end-to-end: a thrown budget-domain error → + * the global AllExceptionsFilter → the canonical envelope + * `{ success, error: { code, message }, requestId }`. The filter is the exact + * class registered as APP_FILTER in AppModule, and the pipe is the exact class + * used by the budget controllers. + */ +describe('budget consumption — error envelopes (real filter + real pipe)', () => { + function mockHost(requestId = 'req-budget-1') { + const json = vi.fn(); + const status = vi.fn(() => ({ json })); + const response = { status }; + const request = { + headers: { 'x-request-id': requestId }, + method: 'POST', + url: '/budgets', + }; + const host = { + switchToHttp: () => ({ getResponse: () => response, getRequest: () => request }), + } as unknown as ArgumentsHost; + return { json, status, host }; + } + + function prismaError(code: string) { + return new Prisma.PrismaClientKnownRequestError('db failed', { + code, + clientVersion: '5.22.0', + }); + } + + const filter = new AllExceptionsFilter(); + + it('maps BudgetExceededException to a 422 BUDGET_EXCEEDED envelope', () => { + const { json, status, host } = mockHost(); + const details = { + budgetId: 'budget-1', + limit: '1000.0000000', + spent: '950.0000000', + attempted: 100, + }; + + filter.catch(new BudgetExceededException('Transaction would exceed the budget limit', details), host); + + expect(status).toHaveBeenCalledWith(422); + expect(json).toHaveBeenCalledWith({ + success: false, + error: { code: 'BUDGET_EXCEEDED', message: 'Transaction would exceed the budget limit', details }, + requestId: 'req-budget-1', + }); + }); + + it('maps ConflictException to a 409 CONFLICT envelope', () => { + const { json, status, host } = mockHost('req-budget-2'); + + filter.catch(new ConflictException('Budget not found'), host); + + expect(status).toHaveBeenCalledWith(409); + expect(json).toHaveBeenCalledWith({ + success: false, + error: { code: 'CONFLICT', message: 'Budget not found' }, + requestId: 'req-budget-2', + }); + }); + + it('maps a Zod validation failure into a 422 VALIDATION_ERROR envelope', () => { + const { json, status, host } = mockHost('req-budget-3'); + const pipe = new ZodValidationPipe(createBudgetSchema); + + // 8 decimal places exceed the platform's 7-dp precision limit. + let thrown: unknown; + try { + pipe.transform({ name: 'Q3', limitAmount: '1000.00000001' }, {} as never); + } catch (error) { + thrown = error; + } + expect(thrown).toBeDefined(); + + filter.catch(thrown, host); + + expect(status).toHaveBeenCalledWith(422); + const body = json.mock.calls[0][0] as { + success: boolean; + error: { code: string; details: Array<{ path: string }> }; + requestId: string; + }; + expect(body.success).toBe(false); + expect(body.error.code).toBe('VALIDATION_ERROR'); + expect(body.error.details).toEqual( + expect.arrayContaining([expect.objectContaining({ path: 'limitAmount' })]), + ); + expect(body.requestId).toBe('req-budget-3'); + }); + + it('maps Prisma P2002 (unique violation) to a 409 CONFLICT envelope', () => { + const { json, status, host } = mockHost('req-budget-4'); + + filter.catch(prismaError('P2002'), host); + + expect(status).toHaveBeenCalledWith(409); + expect(json).toHaveBeenCalledWith({ + success: false, + error: { + code: 'CONFLICT', + message: 'A resource with these unique attributes already exists', + }, + requestId: 'req-budget-4', + }); + }); + + it('maps Prisma P2025 (record not found) to a 404 NOT_FOUND envelope', () => { + const { json, status, host } = mockHost('req-budget-5'); + + filter.catch(prismaError('P2025'), host); + + expect(status).toHaveBeenCalledWith(404); + expect(json).toHaveBeenCalledWith({ + success: false, + error: { code: 'NOT_FOUND', message: 'Resource not found' }, + requestId: 'req-budget-5', + }); + }); + + it('maps unknown Prisma codes (e.g. P1000) to a typed 400 envelope, never leaking the raw error', () => { + const { json, status, host } = mockHost('req-budget-6'); + + filter.catch(prismaError('P1000'), host); + + expect(status).toHaveBeenCalledWith(400); + expect(json).toHaveBeenCalledWith({ + success: false, + error: { code: 'BAD_REQUEST', message: 'Database request could not be processed' }, + requestId: 'req-budget-6', + }); + }); +}); diff --git a/src/modules/transactions/transaction.service.ts b/src/modules/transactions/transaction.service.ts index a785f6a..acbf1a7 100644 --- a/src/modules/transactions/transaction.service.ts +++ b/src/modules/transactions/transaction.service.ts @@ -97,34 +97,46 @@ export class TransactionService { { actorId }, ); - // 5. Budget headroom (no mutation yet). + // 5. Reserve the budget headroom (authoritative — persisted atomically under + // a lock, so concurrent agent requests cannot overspend a single budget). if (input.budgetId) { - await this.budgets.assertWithinBudget(organizationId, input.budgetId, amount); + await this.budgets.reserve(organizationId, input.budgetId, amount); } const requiresApproval = policyResult.requiresApproval || !assessment.canAutoExecute; - // 6. Persist the transaction row. - const transaction = await this.repository.create({ - organization: { connect: { id: organizationId } }, - wallet: { connect: { id: wallet.id } }, - ...(input.agentId ? { agent: { connect: { id: input.agentId } } } : {}), - ...(policyResult.matchedPolicyId - ? { policy: { connect: { id: policyResult.matchedPolicyId } } } - : {}), - ...(input.budgetId ? { budget: { connect: { id: input.budgetId } } } : {}), - asset: input.asset, - amount: new Decimal(input.amount), - senderAddress: wallet.stellarAddress, - recipientAddress: input.recipientAddress, - memo: input.memo, - purpose: input.purpose, - status: TransactionStatus.DRAFT, - riskScore: assessment.score, - riskBand: assessment.band, - requiresApproval, - metadata: input.metadata as Prisma.InputJsonValue, - }); + // 6. Persist the transaction row. If persistence fails, return the reserved + // headroom so the budget is not silently eaten. + let transaction: Transaction; + try { + transaction = await this.repository.create({ + organization: { connect: { id: organizationId } }, + wallet: { connect: { id: wallet.id } }, + ...(input.agentId ? { agent: { connect: { id: input.agentId } } } : {}), + ...(policyResult.matchedPolicyId + ? { policy: { connect: { id: policyResult.matchedPolicyId } } } + : {}), + ...(input.budgetId ? { budget: { connect: { id: input.budgetId } } } : {}), + asset: input.asset, + amount: new Decimal(input.amount), + senderAddress: wallet.stellarAddress, + recipientAddress: input.recipientAddress, + memo: input.memo, + purpose: input.purpose, + status: TransactionStatus.DRAFT, + riskScore: assessment.score, + riskBand: assessment.band, + requiresApproval, + metadata: input.metadata as Prisma.InputJsonValue, + }); + } catch (error) { + if (input.budgetId) { + await this.budgets + .release(organizationId, input.budgetId, amount) + .catch(() => undefined); + } + throw error; + } await this.eventBus.emit( DomainEventName.TransactionCreated, { transactionId: transaction.id, amount: input.amount, riskBand: assessment.band }, @@ -171,6 +183,7 @@ export class TransactionService { { organizationId, actorId, aggregateType: 'transaction', aggregateId: tx.id }, ); + let fundsMoved = false; try { const result = await this.stellar.submitPayment({ sourceAddress: wallet.stellarAddress, @@ -180,6 +193,7 @@ export class TransactionService { memo: tx.memo ?? undefined, network: toNetworkName(wallet.network), }); + fundsMoved = result.successful; const completed = await this.repository.update(tx.id, { status: result.successful ? TransactionStatus.COMPLETED : TransactionStatus.FAILED, @@ -188,6 +202,8 @@ export class TransactionService { }); if (result.successful && tx.budgetId) { + // Settles the reservation booked at creation — spend was already + // reserved, so this only records the event and warns near the cap. await this.budgets.consume(organizationId, tx.budgetId, Number(tx.amount)); } @@ -204,6 +220,12 @@ export class TransactionService { `Transaction ${tx.id} failed to submit: ${(error as Error).message}`, ); await this.repository.update(tx.id, { status: TransactionStatus.FAILED }); + if (tx.budgetId && !fundsMoved) { + // Funds never moved — return the reserved headroom to the budget. + await this.budgets + .release(organizationId, tx.budgetId, Number(tx.amount)) + .catch(() => undefined); + } await this.eventBus.emit( DomainEventName.TransactionFailed, { transactionId: tx.id, reason: (error as Error).message }, @@ -268,6 +290,12 @@ export class TransactionService { throw new ConflictException('Only draft or pending transactions can be cancelled'); } const cancelled = await this.repository.update(id, { status: TransactionStatus.CANCELLED }); + if (tx.budgetId) { + // A draft/pending transaction never moved funds — release the reservation. + await this.budgets + .release(organizationId, tx.budgetId, Number(tx.amount)) + .catch(() => undefined); + } await this.eventBus.emit( DomainEventName.TransactionCancelled, { transactionId: id },