From 95d9a2d4f2effe51684cf96dff2d49ddd4532cc8 Mon Sep 17 00:00:00 2001 From: orebams Date: Tue, 29 Sep 2026 21:37:35 +0100 Subject: [PATCH 1/2] feat(api): paginate, sort, and filter collection endpoints (#475) --- docs/API_REFERENCE.md | 42 ++++++++++++- docs/openapi.yaml | 58 +++++++++++++---- src/approvals/service.ts | 22 +++++-- src/routes/agent-decisions.ts | 14 +++-- src/routes/alerts.ts | 58 +++++++++++++++-- src/routes/approvals.ts | 48 +++++++++++--- src/routes/keys.ts | 76 ++++++++++++++++------- src/routes/portfolio.ts | 66 +++++++++++++++----- src/routes/recurring-deposits.ts | 46 ++++++++++++-- src/routes/sessions.ts | 62 +++++++++++++----- src/routes/transactions.ts | 56 ++++++++++++++--- src/routes/webhooks.ts | 71 +++++++++++++++------ src/utils/pagination.ts | 36 +++++++++-- tests/unit/routes/agent-decisions.test.ts | 32 ++++++++++ tests/unit/utils/pagination.test.ts | 40 ++++++++++++ 15 files changed, 598 insertions(+), 129 deletions(-) create mode 100644 tests/unit/utils/pagination.test.ts diff --git a/docs/API_REFERENCE.md b/docs/API_REFERENCE.md index 1387c8a0..8b470411 100644 --- a/docs/API_REFERENCE.md +++ b/docs/API_REFERENCE.md @@ -15,6 +15,35 @@ Comprehensive reference for all backend endpoints defined in src/routes. - Every response includes an `X-API-Version: 1` header. - Breaking changes introduce a new major version (`/api/v2`); deprecated versions are supported for a minimum of 6 months before the announced `Sunset` date. The full policy lives in [`docs/api-versioning.md`](api-versioning.md). +## List Queries + +Collection endpoints use bounded page-based pagination. `page` is 1-based and +defaults to `1`; `limit` defaults to `20` and is capped at `50`. The equivalent +SQL offset is `(page - 1) * limit`. List responses include `page`, `limit`, +`total`, `totalPages`, `hasNext`, and `hasPrevious`; an empty result has +`totalPages: 0` and both navigation flags set to `false`. + +Where supported, `sortBy` is restricted to the fields documented for that +resource and `sortOrder` is `asc` or `desc` (default `desc`). The server applies +a stable ID tie-breaker. Filters are resource-specific and are applied before +both counting and fetching. Invalid page, limit, sort, or filter values return +`400` rather than being silently ignored. + +This contract is used by transactions, portfolio positions, recurring deposit +plans, alert rules, webhooks, API keys, sessions, approvals, and agent decisions. + +Supported list controls: + +- Transactions: filter by `type`, `status`, `protocolName`, and inclusive `from`/`to` creation dates; sort by `createdAt`, `updatedAt`, or `amount`. +- Portfolio positions: filter by `status`, `protocolName`, and `assetSymbol`; sort by `openedAt`, `updatedAt`, `currentValue`, or `yieldEarned`. Portfolio summary totals remain account-wide. +- Recurring deposits: filter by `status`, `cadence`, and `assetSymbol`; sort by `createdAt`, `nextRunAt`, or `amount`. +- Alert rules: filter by `isActive`, `metric`, and `protocolName`; sort by `createdAt`, `updatedAt`, or `threshold`. +- Webhooks: filter by `isActive` and `event`; sort by `createdAt` or `updatedAt`. +- API keys: filter by `revoked`; sort by `createdAt`, `lastUsedAt`, or `expiresAt`. +- Sessions: sort by `lastSeenAt`, `createdAt`, or `expiresAt`. +- Approvals: filter by `status`; sort by `requestedAt` or `executedAt`. +- Agent decisions: filter by `outcome`, `fromProtocol`, and inclusive `from`/`to` creation dates; sort by `createdAt`, `outcome`, or `fromProtocol`. + ## Authentication and Authorization - Public endpoints: GET /health, POST /api/auth/challenge, POST /api/auth/verify, GET /api/whatsapp/webhook, POST /api/whatsapp/webhook, GET /api/vault/state, GET /api/protocols/rates, GET /api/protocols/agent/status, GET /api/agent/status @@ -508,17 +537,26 @@ Response 404: - userId: uuid string - Query params: - page: int >= 1, default 1 - - limit: int between 1 and 50, default 5 + - limit: int between 1 and 50, default 20 + - type: optional transaction type + - status: optional transaction status + - protocolName: optional exact protocol match + - from, to: optional inclusive ISO-8601 bounds on createdAt + - sortBy: createdAt, updatedAt, or amount (default createdAt) + - sortOrder: asc or desc (default desc) - Request body: none Example request: -GET /api/transactions/550e8400-e29b-41d4-a716-446655440002?page=2&limit=10 +GET /api/transactions/550e8400-e29b-41d4-a716-446655440002?page=2&limit=10&status=CONFIRMED&sortBy=amount&sortOrder=desc Response 200: { "page": 2, "limit": 10, "total": 20, +"totalPages": 2, +"hasNext": false, +"hasPrevious": true, "transactions": [ { "id": "tx-id-1", diff --git a/docs/openapi.yaml b/docs/openapi.yaml index c64a388c..925cb96e 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -70,6 +70,28 @@ components: window. Rows/responses with this flag set are excluded from leaderboard rankings. Never silently serves a truncated window under the requested label. + PaginationMeta: + type: object + required: [page, limit, total, totalPages, hasNext, hasPrevious] + properties: + page: + type: integer + minimum: 1 + limit: + type: integer + minimum: 1 + maximum: 50 + total: + type: integer + minimum: 0 + totalPages: + type: integer + minimum: 0 + hasNext: + type: boolean + hasPrevious: + type: boolean + NullableDecimal: type: number format: double @@ -1888,25 +1910,35 @@ paths: type: integer minimum: 1 maximum: 50 - default: 10 + default: 20 + - name: sortBy + in: query + required: false + schema: + type: string + enum: [createdAt, outcome, fromProtocol] + default: createdAt + - name: sortOrder + in: query + required: false + schema: + type: string + enum: [asc, desc] + default: desc responses: '200': description: Paginated decision listing (user-projected). content: application/json: schema: - type: object - properties: - page: - type: integer - limit: - type: integer - total: - type: integer - decisions: - type: array - items: - $ref: '#/components/schemas/RebalanceDecision' + allOf: + - $ref: '#/components/schemas/PaginationMeta' + - type: object + properties: + decisions: + type: array + items: + $ref: '#/components/schemas/RebalanceDecision' '401': $ref: '#/components/responses/Unauthorized' diff --git a/src/approvals/service.ts b/src/approvals/service.ts index 15339062..1ee0ed51 100644 --- a/src/approvals/service.ts +++ b/src/approvals/service.ts @@ -33,7 +33,7 @@ import db from '../db' import { logger } from '../utils/logger' import { AppError } from '../utils/errors' import { dispatchWebhookEvent } from '../services/webhookDispatcher' -import { getPaginationParams } from '../utils/pagination' +import { buildPaginationMeta, getPaginationParams } from '../utils/pagination' import { runApprovedPayload, type ApprovalPayload } from './executors' type Db = typeof db | Prisma.TransactionClient @@ -410,18 +410,29 @@ export async function cancel( export async function listApprovalRequestsForUser( userId: string, - query: { page?: unknown; limit?: unknown }, + query: { + page?: unknown + limit?: unknown + status?: + 'PENDING' | 'APPROVED' | 'EXECUTED' | 'REJECTED' | 'EXPIRED' | 'CANCELLED' + sortBy?: 'requestedAt' | 'executedAt' + sortOrder?: 'asc' | 'desc' + }, database: Db = db ): Promise<{ requests: ApprovalRequest[] page: number limit: number total: number + totalPages: number + hasNext: boolean + hasPrevious: boolean }> { const { page, limit, skip } = getPaginationParams(query) const parentIds = await canSeeRequestsForPolicyOwner(userId, database) const where: Prisma.ApprovalRequestWhereInput = { + ...(query.status ? { status: query.status } : {}), OR: [ { policy: { principalUserId: userId } }, ...(parentIds.length > 0 @@ -433,14 +444,17 @@ export async function listApprovalRequestsForUser( const [requests, total] = await Promise.all([ database.approvalRequest.findMany({ where, - orderBy: { requestedAt: 'desc' }, + orderBy: [ + { [query.sortBy ?? 'requestedAt']: query.sortOrder ?? 'desc' }, + { id: 'desc' }, + ], skip, take: limit, }), database.approvalRequest.count({ where }), ]) - return { requests, page, limit, total } + return { requests, ...buildPaginationMeta(page, limit, total) } } export async function getVisibleRequestDetail( diff --git a/src/routes/agent-decisions.ts b/src/routes/agent-decisions.ts index c2936162..01dea0aa 100644 --- a/src/routes/agent-decisions.ts +++ b/src/routes/agent-decisions.ts @@ -17,6 +17,7 @@ import db from '../db' import { requireAuth } from '../middleware/authenticate' import { logger } from '../utils/logger' import { sendNotFound } from '../utils/errors' +import { buildPaginationMeta } from '../utils/pagination' const router = Router() @@ -87,7 +88,9 @@ const listQuerySchema = z.object({ from: z.string().datetime({ offset: true }).optional(), to: z.string().datetime({ offset: true }).optional(), page: z.coerce.number().int().min(1).default(1), - limit: z.coerce.number().int().min(1).max(50).default(10), + limit: z.coerce.number().int().min(1).max(50).default(20), + sortBy: z.enum(['createdAt', 'outcome', 'fromProtocol']).default('createdAt'), + sortOrder: z.enum(['asc', 'desc']).default('desc'), }) const idParamSchema = z.object({ @@ -107,7 +110,8 @@ router.get('/', requireAuth, async (req: Request, res: Response) => { .json({ error: 'Validation failed', details: parsed.error.flatten() }) } - const { outcome, fromProtocol, from, to, page, limit } = parsed.data + const { outcome, fromProtocol, from, to, page, limit, sortBy, sortOrder } = + parsed.data const skip = (page - 1) * limit const where: any = { @@ -126,7 +130,7 @@ router.get('/', requireAuth, async (req: Request, res: Response) => { (db as any).rebalanceDecision.count({ where }), (db as any).rebalanceDecision.findMany({ where, - orderBy: { createdAt: 'desc' }, + orderBy: [{ [sortBy]: sortOrder }, { id: 'desc' }], skip, take: limit, }), @@ -153,9 +157,7 @@ router.get('/', requireAuth, async (req: Request, res: Response) => { ) return res.status(200).json({ - page, - limit, - total, + ...buildPaginationMeta(page, limit, total), decisions: data, }) } catch (error) { diff --git a/src/routes/alerts.ts b/src/routes/alerts.ts index c978dc42..6c2da69b 100644 --- a/src/routes/alerts.ts +++ b/src/routes/alerts.ts @@ -1,4 +1,5 @@ import { Router, Request, Response } from 'express' +import { z } from 'zod' import db from '../db' import { requireAuth, enforceUserAccess } from '../middleware/authenticate' import { requireScope } from '../middleware/apiKeyAuth' @@ -11,6 +12,11 @@ import { alertUserParamSchema, compositeAlertRuleSchema, } from '../validators/alert-validators' +import { + buildPaginationMeta, + getPaginationParams, + paginationSchema, +} from '../utils/pagination' const router = Router() @@ -98,17 +104,57 @@ router.post( router.get( '/:userId', validate({ params: alertUserParamSchema }), + validate({ + query: paginationSchema.extend({ + isActive: z.enum(['true', 'false']).optional(), + metric: z + .enum([ + 'PROTOCOL_APY', + 'PORTFOLIO_VALUE', + 'POSITION_DRAWDOWN', + 'DRIFT', + 'VOLATILITY_REGIME', + 'ANOMALY', + ]) + .optional(), + protocolName: z.string().trim().min(1).max(100).optional(), + sortBy: z + .enum(['createdAt', 'updatedAt', 'threshold']) + .default('createdAt'), + sortOrder: z.enum(['asc', 'desc']).default('desc'), + }), + }), enforceUserAccess, async (req: Request, res: Response) => { const userId = req.params.userId as string + const { page, limit, skip } = getPaginationParams(req.query) + const query = req.query as { + isActive?: 'true' | 'false' + metric?: string + protocolName?: string + sortBy: 'createdAt' | 'updatedAt' | 'threshold' + sortOrder: 'asc' | 'desc' + } - const rules = await (db as any).alertRule.findMany({ - where: { userId }, - select: alertSelect, - orderBy: { createdAt: 'desc' }, - }) + const where: any = { userId } + if (query.isActive !== undefined) where.isActive = query.isActive === 'true' + if (query.metric) where.metric = query.metric + if (query.protocolName) where.protocolName = query.protocolName + + const [total, rules] = await Promise.all([ + (db as any).alertRule.count({ where }), + (db as any).alertRule.findMany({ + where, + select: alertSelect, + orderBy: [{ [query.sortBy]: query.sortOrder }, { id: 'desc' }], + skip, + take: limit, + }), + ]) - return res.status(200).json({ rules }) + return res + .status(200) + .json({ ...buildPaginationMeta(page, limit, total), rules }) } ) diff --git a/src/routes/approvals.ts b/src/routes/approvals.ts index 19c6cf22..91d8b179 100644 --- a/src/routes/approvals.ts +++ b/src/routes/approvals.ts @@ -1,4 +1,6 @@ import { Router, Request, Response } from 'express' +import { z } from 'zod' +import { paginationSchema } from '../utils/pagination' import { requireAuth } from '../middleware/authenticate' import { validate } from '../middleware/validate' import { idempotent } from '../middleware/idempotency' @@ -14,6 +16,21 @@ import { const router = Router() +const approvalListQuerySchema = paginationSchema.extend({ + status: z + .enum([ + 'PENDING', + 'APPROVED', + 'EXECUTED', + 'REJECTED', + 'EXPIRED', + 'CANCELLED', + ]) + .optional(), + sortBy: z.enum(['requestedAt', 'executedAt']).default('requestedAt'), + sortOrder: z.enum(['asc', 'desc']).default('desc'), +}) + function handleServiceError(res: Response, err: unknown, action: string) { if (err instanceof AppError) { return sendError(res, err.statusCode, err.message) @@ -25,17 +42,28 @@ function handleServiceError(res: Response, err: unknown, action: string) { } // ── GET / — requests affecting the caller (as principal or eligible approver) ── -router.get('/', requireAuth, async (req: Request, res: Response) => { - try { - const result = await listApprovalRequestsForUser(req.auth!.userId, { - page: req.query.page, - limit: req.query.limit, - }) - res.json(result) - } catch (err) { - handleServiceError(res, err, 'List') +router.get( + '/', + requireAuth, + validate({ query: approvalListQuerySchema }), + async (req: Request, res: Response) => { + try { + const query = req.query as unknown as z.infer< + typeof approvalListQuerySchema + > + const result = await listApprovalRequestsForUser(req.auth!.userId, { + page: query.page, + limit: query.limit, + status: query.status, + sortBy: query.sortBy, + sortOrder: query.sortOrder, + }) + res.json(result) + } catch (err) { + handleServiceError(res, err, 'List') + } } -}) +) // ── GET /:id — full request + decisions ───────────────────────────────────── router.get('/:id', requireAuth, async (req: Request, res: Response) => { diff --git a/src/routes/keys.ts b/src/routes/keys.ts index 4e1cb3ae..84b07f6b 100644 --- a/src/routes/keys.ts +++ b/src/routes/keys.ts @@ -16,6 +16,11 @@ import { config } from '../config' import { logger } from '../utils/logger' import { validateUserScopes, USER_SCOPES, type UserScope } from '../auth/scopes' import { publishUserEvent } from '../events/publisher' +import { + buildPaginationMeta, + getPaginationParams, + paginationSchema, +} from '../utils/pagination' const router = Router() const prisma = db as any @@ -133,27 +138,56 @@ router.post( ) /** GET /api/v1/keys — list metadata (no secrets). */ -router.get('/', async (req: Request, res: Response) => { - const userId = req.auth!.userId - const keys = await prisma.userApiKey.findMany({ - where: { userId }, - select: { - id: true, - name: true, - scopes: true, - ipAllowlist: true, - rateLimitPerMin: true, - allowWithdrawals: true, - lastUsedAt: true, - lastUsedIp: true, - expiresAt: true, - revokedAt: true, - createdAt: true, - }, - orderBy: { createdAt: 'desc' }, - }) - return res.status(200).json({ keys }) -}) +router.get( + '/', + validate({ + query: paginationSchema.extend({ + revoked: z.enum(['true', 'false']).optional(), + sortBy: z + .enum(['createdAt', 'lastUsedAt', 'expiresAt']) + .default('createdAt'), + sortOrder: z.enum(['asc', 'desc']).default('desc'), + }), + }), + async (req: Request, res: Response) => { + const userId = req.auth!.userId + const { page, limit, skip } = getPaginationParams(req.query) + const query = req.query as { + revoked?: 'true' | 'false' + sortBy: 'createdAt' | 'lastUsedAt' | 'expiresAt' + sortOrder: 'asc' | 'desc' + } + const where: any = { userId } + if (query.revoked !== undefined) { + where.revokedAt = query.revoked === 'true' ? { not: null } : null + } + const [total, keys] = await Promise.all([ + prisma.userApiKey.count({ where }), + prisma.userApiKey.findMany({ + where, + select: { + id: true, + name: true, + scopes: true, + ipAllowlist: true, + rateLimitPerMin: true, + allowWithdrawals: true, + lastUsedAt: true, + lastUsedIp: true, + expiresAt: true, + revokedAt: true, + createdAt: true, + }, + orderBy: [{ [query.sortBy]: query.sortOrder }, { id: 'desc' }], + skip, + take: limit, + }), + ]) + return res + .status(200) + .json({ ...buildPaginationMeta(page, limit, total), keys }) + } +) /** DELETE /api/v1/keys/:id — revoke a key. */ router.delete( diff --git a/src/routes/portfolio.ts b/src/routes/portfolio.ts index bb2fe882..a5340cee 100644 --- a/src/routes/portfolio.ts +++ b/src/routes/portfolio.ts @@ -34,6 +34,11 @@ import { } from '../tax/report' import { AccountingMethod } from '@prisma/client' import { toCsv } from '../utils/csv' +import { + buildPaginationMeta, + getPaginationParams, + paginationSchema, +} from '../utils/pagination' import goalsRouter from './goals' const router = Router() @@ -42,12 +47,24 @@ const router = Router() // is never captured as a userId. router.use('/goals', goalsRouter) -const portfolioSchema = z.object({ +const portfolioParamsSchema = z.object({ params: z.object({ userId: z.string().uuid(), }), }) +const portfolioSchema = portfolioParamsSchema.extend({ + query: paginationSchema.extend({ + status: z.enum(['ACTIVE', 'CLOSED', 'LIQUIDATED']).optional(), + protocolName: z.string().trim().min(1).max(100).optional(), + assetSymbol: z.string().trim().min(1).max(20).optional(), + sortBy: z + .enum(['openedAt', 'updatedAt', 'currentValue', 'yieldEarned']) + .default('openedAt'), + sortOrder: z.enum(['asc', 'desc']).default('desc'), + }), +}) + const taxReportSchema = z.object({ params: z.object({ userId: z.string().uuid(), @@ -202,24 +219,43 @@ router.get( return sendNotFound(res, 'User') } - const userPositions = await db.position.findMany({ - where: { userId }, - }) - - const totalBalance = userPositions.reduce((sum: number, position: any) => { - return sum + Number(position.currentValue) - }, 0) - const totalEarnings = userPositions.reduce((sum: number, position: any) => { - return sum + Number(position.yieldEarned) - }, 0) - const activePositions = userPositions.filter( - (p: any) => p.status === 'ACTIVE' - ).length + const { page, limit, skip } = getPaginationParams(req.query) + const query = req.query as { + status?: 'ACTIVE' | 'CLOSED' | 'LIQUIDATED' + protocolName?: string + assetSymbol?: string + sortBy: 'openedAt' | 'updatedAt' | 'currentValue' | 'yieldEarned' + sortOrder: 'asc' | 'desc' + } + const where: any = { userId } + if (query.status) where.status = query.status + if (query.protocolName) where.protocolName = query.protocolName + if (query.assetSymbol) where.assetSymbol = query.assetSymbol + + const [userPositions, total, portfolioTotals, activePositions] = + await Promise.all([ + db.position.findMany({ + where, + orderBy: [{ [query.sortBy]: query.sortOrder }, { id: 'desc' }], + skip, + take: limit, + }), + db.position.count({ where }), + db.position.aggregate({ + where: { userId }, + _sum: { currentValue: true, yieldEarned: true }, + }), + db.position.count({ where: { userId, status: 'ACTIVE' } }), + ]) + + const totalBalance = Number(portfolioTotals._sum.currentValue ?? 0) + const totalEarnings = Number(portfolioTotals._sum.yieldEarned ?? 0) const positions = userPositions.map(mapPositionToResponse) return res.status(200).json({ userId: user.id, + ...buildPaginationMeta(page, limit, total), totalBalance, totalEarnings, activePositions, @@ -283,7 +319,7 @@ router.get( '/:userId/earnings', requireAuth, enforceUserAccess, - validate(portfolioSchema), + validate(portfolioParamsSchema), async (req: Request, res: Response) => { const userId = req.params.userId as string const user = await db.user.findUnique({ diff --git a/src/routes/recurring-deposits.ts b/src/routes/recurring-deposits.ts index 584d4803..476b6220 100644 --- a/src/routes/recurring-deposits.ts +++ b/src/routes/recurring-deposits.ts @@ -1,4 +1,5 @@ import { Router, Request, Response } from 'express' +import { z } from 'zod' import { requireAuth, enforceUserAccess } from '../middleware/authenticate' import { requireScope } from '../middleware/apiKeyAuth' import { idempotent } from '../middleware/idempotency' @@ -11,6 +12,11 @@ import { } from '../validators/recurring-deposit-validators' import db from '../db' import { addCadence } from '../utils/cadence' +import { + buildPaginationMeta, + getPaginationParams, + paginationSchema, +} from '../utils/pagination' const router = Router() @@ -70,15 +76,43 @@ router.get( '/by-user/:userId', requireAuth, enforceUserAccess, + validate({ + params: z.object({ userId: z.string().uuid() }), + query: paginationSchema.extend({ + status: z.enum(['ACTIVE', 'PAUSED', 'CANCELLED']).optional(), + cadence: z.enum(['WEEKLY', 'BIWEEKLY', 'MONTHLY']).optional(), + assetSymbol: z.string().trim().min(1).max(20).optional(), + sortBy: z.enum(['createdAt', 'nextRunAt', 'amount']).default('createdAt'), + sortOrder: z.enum(['asc', 'desc']).default('desc'), + }), + }), async (req: Request, res: Response) => { const { userId } = req.params + const { page, limit, skip } = getPaginationParams(req.query) + const query = req.query as { + status?: string + cadence?: string + assetSymbol?: string + sortBy: 'createdAt' | 'nextRunAt' | 'amount' + sortOrder: 'asc' | 'desc' + } - const plans = await db.recurringDepositPlan.findMany({ - where: { userId }, - orderBy: { createdAt: 'desc' }, - }) - - return res.json({ plans }) + const where: any = { userId } + if (query.status) where.status = query.status + if (query.cadence) where.cadence = query.cadence + if (query.assetSymbol) where.assetSymbol = query.assetSymbol + + const [total, plans] = await Promise.all([ + db.recurringDepositPlan.count({ where }), + db.recurringDepositPlan.findMany({ + where, + orderBy: [{ [query.sortBy]: query.sortOrder }, { id: 'desc' }], + skip, + take: limit, + }), + ]) + + return res.json({ ...buildPaginationMeta(page, limit, total), plans }) } ) diff --git a/src/routes/sessions.ts b/src/routes/sessions.ts index da3c884c..e2c53ee2 100644 --- a/src/routes/sessions.ts +++ b/src/routes/sessions.ts @@ -5,6 +5,11 @@ import { requireAuth } from '../middleware/authenticate' import { requireSessionAuth } from '../middleware/apiKeyAuth' import { validate } from '../middleware/validate' import { sendNotFound } from '../utils/errors' +import { + buildPaginationMeta, + getPaginationParams, + paginationSchema, +} from '../utils/pagination' import { maskIpAddress } from '../utils/geoip' import { revokeSession } from '../services/refresh-token.service' import { stellarVerification } from '../utils/stellar/stellar-verification' @@ -44,22 +49,47 @@ function formatSession( } /** GET /api/v1/sessions */ -router.get('/', async (req: Request, res: Response) => { - const userId = req.auth!.userId - const currentSessionId = req.auth!.sessionId - const showFullIp = req.query.fullIp === 'true' - - const sessions = await prisma.session.findMany({ - where: { userId, revokedAt: null, expiresAt: { gt: new Date() } }, - orderBy: { lastSeenAt: 'desc' }, - }) - - return res.status(200).json({ - sessions: sessions.map((s: Record) => - formatSession(s, currentSessionId, showFullIp) - ), - }) -}) +router.get( + '/', + validate({ + query: paginationSchema.extend({ + fullIp: z.enum(['true', 'false']).default('false'), + sortBy: z + .enum(['lastSeenAt', 'createdAt', 'expiresAt']) + .default('lastSeenAt'), + sortOrder: z.enum(['asc', 'desc']).default('desc'), + }), + }), + async (req: Request, res: Response) => { + const userId = req.auth!.userId + const currentSessionId = req.auth!.sessionId + const { page, limit, skip } = getPaginationParams(req.query) + const query = req.query as { + fullIp: 'true' | 'false' + sortBy: 'lastSeenAt' | 'createdAt' | 'expiresAt' + sortOrder: 'asc' | 'desc' + } + const showFullIp = query.fullIp === 'true' + const where = { userId, revokedAt: null, expiresAt: { gt: new Date() } } + + const [total, sessions] = await Promise.all([ + prisma.session.count({ where }), + prisma.session.findMany({ + where, + orderBy: [{ [query.sortBy]: query.sortOrder }, { id: 'desc' }], + skip, + take: limit, + }), + ]) + + return res.status(200).json({ + ...buildPaginationMeta(page, limit, total), + sessions: sessions.map((s: Record) => + formatSession(s, currentSessionId, showFullIp) + ), + }) + } +) /** PATCH /api/v1/sessions/:id — set label */ router.patch( diff --git a/src/routes/transactions.ts b/src/routes/transactions.ts index 19fb271b..89ce71fd 100644 --- a/src/routes/transactions.ts +++ b/src/routes/transactions.ts @@ -3,7 +3,11 @@ import { z } from 'zod' import db from '../db' import { requireAuth, enforceUserAccess } from '../middleware/authenticate' import { validate } from '../middleware/validate' -import { paginationSchema, getPaginationParams } from '../utils/pagination' +import { + buildPaginationMeta, + getPaginationParams, + paginationSchema, +} from '../utils/pagination' import { mapTransactionToResponse } from '../utils/api-formatters' import { sendNotFound } from '../utils/errors' import { @@ -22,7 +26,26 @@ const listSchema = z.object({ params: z.object({ userId: z.string().uuid(), }), - query: paginationSchema, + query: paginationSchema.extend({ + type: z + .enum([ + 'DEPOSIT', + 'WITHDRAWAL', + 'YIELD_CLAIM', + 'REBALANCE', + 'SWAP', + 'REFERRAL_REWARD', + 'INBOUND_TRANSFER', + 'CLAIMABLE_BALANCE_CLAIM', + ]) + .optional(), + status: z.enum(['PENDING', 'CONFIRMED', 'FAILED', 'CANCELLED']).optional(), + protocolName: z.string().trim().min(1).max(100).optional(), + from: z.string().datetime({ offset: true }).optional(), + to: z.string().datetime({ offset: true }).optional(), + sortBy: z.enum(['createdAt', 'updatedAt', 'amount']).default('createdAt'), + sortOrder: z.enum(['asc', 'desc']).default('desc'), + }), }) // ─── Router ─────────────────────────────────────────────────────────────────── @@ -68,6 +91,15 @@ router.get( async (req: Request, res: Response) => { const userId = String(req.params.userId) const { page, limit, skip } = getPaginationParams(req.query) + const query = req.query as { + type?: string + status?: string + protocolName?: string + from?: string + to?: string + sortBy: 'createdAt' | 'updatedAt' | 'amount' + sortOrder: 'asc' | 'desc' + } const user = await db.user.findUnique({ where: { id: userId }, @@ -75,11 +107,21 @@ router.get( }) if (!user) return sendNotFound(res, 'User') + const where: any = { userId } + if (query.type) where.type = query.type + if (query.status) where.status = query.status + if (query.protocolName) where.protocolName = query.protocolName + if (query.from || query.to) { + where.createdAt = {} + if (query.from) where.createdAt.gte = new Date(query.from) + if (query.to) where.createdAt.lte = new Date(query.to) + } + const [total, transactions] = await Promise.all([ - db.transaction.count({ where: { userId } }), + db.transaction.count({ where }), db.transaction.findMany({ - where: { userId }, - orderBy: { createdAt: 'desc' }, + where, + orderBy: [{ [query.sortBy]: query.sortOrder }, { id: 'desc' }], skip, take: limit, }), @@ -88,9 +130,7 @@ router.get( const items = transactions.map(mapTransactionToResponse) return res.status(200).json({ - page, - limit, - total, + ...buildPaginationMeta(page, limit, total), transactions: items, whatsappReply: formatTransactionsReply({ page, diff --git a/src/routes/webhooks.ts b/src/routes/webhooks.ts index cfb20794..d83deb95 100644 --- a/src/routes/webhooks.ts +++ b/src/routes/webhooks.ts @@ -1,4 +1,5 @@ import { Router, Request, Response } from 'express' +import { z } from 'zod' import db from '../db' import { requireAuth } from '../middleware/authenticate' import { requireScope } from '../middleware/apiKeyAuth' @@ -16,6 +17,11 @@ import { import { getSubscriptionHealth } from '../services/webhookCircuitBreaker' import { replayDeadLetter } from '../services/webhookDispatcher' import { logger } from '../utils/logger' +import { + buildPaginationMeta, + getPaginationParams, + paginationSchema, +} from '../utils/pagination' const router = Router() const prisma = db as any @@ -80,24 +86,53 @@ router.post( ) /** GET /api/webhooks */ -router.get('/', async (req: Request, res: Response) => { - const userId = req.auth!.userId - const subscriptions = await prisma.webhookSubscription.findMany({ - where: { userId }, - select: { - id: true, - url: true, - events: true, - isActive: true, - autoReplay: true, - secretNextActiveAt: true, - createdAt: true, - updatedAt: true, - }, - orderBy: { createdAt: 'desc' }, - }) - return res.status(200).json({ subscriptions }) -}) +router.get( + '/', + validate({ + query: paginationSchema.extend({ + isActive: z.enum(['true', 'false']).optional(), + event: z.string().trim().min(1).max(100).optional(), + sortBy: z.enum(['createdAt', 'updatedAt']).default('createdAt'), + sortOrder: z.enum(['asc', 'desc']).default('desc'), + }), + }), + async (req: Request, res: Response) => { + const userId = req.auth!.userId + const { page, limit, skip } = getPaginationParams(req.query) + const query = req.query as { + isActive?: 'true' | 'false' + event?: string + sortBy: 'createdAt' | 'updatedAt' + sortOrder: 'asc' | 'desc' + } + const where: any = { userId } + if (query.isActive !== undefined) where.isActive = query.isActive === 'true' + if (query.event) where.events = { has: query.event } + + const [total, subscriptions] = await Promise.all([ + prisma.webhookSubscription.count({ where }), + prisma.webhookSubscription.findMany({ + where, + select: { + id: true, + url: true, + events: true, + isActive: true, + autoReplay: true, + secretNextActiveAt: true, + createdAt: true, + updatedAt: true, + }, + orderBy: [{ [query.sortBy]: query.sortOrder }, { id: 'desc' }], + skip, + take: limit, + }), + ]) + return res + .status(200) + .json({ ...buildPaginationMeta(page, limit, total), subscriptions }) + } +) router.get( '/:id', diff --git a/src/utils/pagination.ts b/src/utils/pagination.ts index a2ab23fc..ae2dd00d 100644 --- a/src/utils/pagination.ts +++ b/src/utils/pagination.ts @@ -2,12 +2,40 @@ import { z } from 'zod' export const paginationSchema = z.object({ page: z.coerce.number().int().min(1).default(1), - limit: z.coerce.number().int().min(1).max(50).default(5), + limit: z.coerce.number().int().min(1).max(50).default(20), }) -export function getPaginationParams(query: any) { - const page = Number(query.page) || 1 - const limit = Number(query.limit) || 5 +export interface PaginationMeta { + page: number + limit: number + total: number + totalPages: number + hasNext: boolean + hasPrevious: boolean +} + +export function getPaginationParams(query: { + page?: unknown + limit?: unknown +}) { + const parsed = paginationSchema.parse(query) + const { page, limit } = parsed const skip = (page - 1) * limit return { page, limit, skip } } + +export function buildPaginationMeta( + page: number, + limit: number, + total: number +): PaginationMeta { + const totalPages = Math.ceil(total / limit) + return { + page, + limit, + total, + totalPages, + hasNext: page < totalPages, + hasPrevious: page > 1 && totalPages > 0, + } +} diff --git a/tests/unit/routes/agent-decisions.test.ts b/tests/unit/routes/agent-decisions.test.ts index 825fd5bd..fdd9db29 100644 --- a/tests/unit/routes/agent-decisions.test.ts +++ b/tests/unit/routes/agent-decisions.test.ts @@ -120,6 +120,38 @@ describe('GET /decisions', () => { }) ) }) + + it('returns pagination metadata and applies a whitelisted sort', async () => { + mockCount.mockResolvedValue(0) + mockFindMany.mockResolvedValue([]) + + const res = await request(app) + .get('/decisions?page=2&limit=10&sortBy=outcome&sortOrder=asc') + .set(authHeader()) + + expect(res.status).toBe(200) + expect(res.body).toMatchObject({ + page: 2, + limit: 10, + total: 0, + totalPages: 0, + hasNext: false, + hasPrevious: false, + decisions: [], + }) + expect(mockFindMany).toHaveBeenCalledWith( + expect.objectContaining({ + orderBy: [{ outcome: 'asc' }, { id: 'desc' }], + }) + ) + }) + + it('rejects invalid pagination before querying', async () => { + const res = await request(app).get('/decisions?page=0').set(authHeader()) + + expect(res.status).toBe(400) + expect(mockFindMany).not.toHaveBeenCalled() + }) }) describe('GET /decisions/:id', () => { diff --git a/tests/unit/utils/pagination.test.ts b/tests/unit/utils/pagination.test.ts new file mode 100644 index 00000000..edd8098a --- /dev/null +++ b/tests/unit/utils/pagination.test.ts @@ -0,0 +1,40 @@ +import { + buildPaginationMeta, + getPaginationParams, + paginationSchema, +} from '../../../src/utils/pagination' + +describe('pagination helpers', () => { + it('uses bounded defaults and computes an offset', () => { + expect(getPaginationParams({})).toEqual({ page: 1, limit: 20, skip: 0 }) + expect(getPaginationParams({ page: '3', limit: '25' })).toEqual({ + page: 3, + limit: 25, + skip: 50, + }) + }) + + it('rejects out-of-range pagination requests', () => { + expect(paginationSchema.safeParse({ page: '0' }).success).toBe(false) + expect(paginationSchema.safeParse({ limit: '51' }).success).toBe(false) + }) + + it('returns stable metadata for empty and populated pages', () => { + expect(buildPaginationMeta(1, 20, 0)).toEqual({ + page: 1, + limit: 20, + total: 0, + totalPages: 0, + hasNext: false, + hasPrevious: false, + }) + expect(buildPaginationMeta(2, 10, 25)).toEqual({ + page: 2, + limit: 10, + total: 25, + totalPages: 3, + hasNext: true, + hasPrevious: true, + }) + }) +}) From 150c961a046e81d03389f2bd519a2b9eaec7beb1 Mon Sep 17 00:00:00 2001 From: orebams Date: Wed, 30 Sep 2026 10:46:37 +0100 Subject: [PATCH 2/2] feat(agent): enforce liquidity floor (#541) --- docs/API_REFERENCE.md | 25 ++ docs/LIQUIDITY_FLOOR.md | 50 ++++ .../migration.sql | 8 + .../rollback.sql | 8 + prisma/schema.prisma | 3 + src/agent/loop.ts | 19 +- src/agent/router.ts | 154 ++++++++++- src/agent/scanner.ts | 30 +++ src/agent/tools/actionTools.ts | 8 +- src/agent/types.ts | 2 + src/analytics/liquidity.ts | 77 ++++++ src/analytics/liquidityFloor.ts | 243 +++++++++++++++++ src/analytics/liquiditySnapshot.ts | 168 ++++++++++++ src/index.ts | 2 + src/routes/liquidity-floor.ts | 63 +++++ .../agent/rebalanceCost.integration.test.ts | 48 ++++ tests/unit/analytics/liquidity.test.ts | 80 ++++++ .../analytics/liquidityFloorService.test.ts | 248 ++++++++++++++++++ tests/unit/routes/liquidity-floor.test.ts | 102 +++++++ 19 files changed, 1331 insertions(+), 7 deletions(-) create mode 100644 docs/LIQUIDITY_FLOOR.md create mode 100644 prisma/migrations/20260930200000_agent_liquidity_floor/migration.sql create mode 100644 prisma/migrations/20260930200000_agent_liquidity_floor/rollback.sql create mode 100644 src/analytics/liquidityFloor.ts create mode 100644 src/analytics/liquiditySnapshot.ts create mode 100644 src/routes/liquidity-floor.ts create mode 100644 tests/unit/analytics/liquidity.test.ts create mode 100644 tests/unit/analytics/liquidityFloorService.test.ts create mode 100644 tests/unit/routes/liquidity-floor.test.ts diff --git a/docs/API_REFERENCE.md b/docs/API_REFERENCE.md index 8b470411..43f71909 100644 --- a/docs/API_REFERENCE.md +++ b/docs/API_REFERENCE.md @@ -327,6 +327,31 @@ Notes: ## Portfolio +### GET /api/v1/liquidity-floor + +- Auth: required; returns only the authenticated caller's status. +- Returns the configured USD floor, active USDC balance, amount exitable within + the 0.5% slippage target, yield-eligible amount, shortfall, estimated + restoration time, and liquidity-data availability. +- Only fresh (24-hour) snapshots with explicit zero withdrawal delay and queue + depth count as instantly liquid. Locked positions, pending transactions, + unsupported assets, and missing or stale data do not count. +- `dataAvailable: false` means the agent blocks rebalancing until liquidity data + is available. A floor above total balance is reported as a shortfall. + +### PATCH /api/v1/liquidity-floor + +- Auth: required; updates only the authenticated caller's floor. +- Request body: `{ "floorUsd": 500 }` sets a $500 minimum; `null` clears it; + zero disables reserve sizing. +- The floor takes precedence over goals and strategy allocation. Restoration + uses shortest known exits first and retains existing cost/payback gates; it is + a target, not an instantaneous guarantee. + +The Stellar DEX collector derives conservative depth from Horizon reserves. +Blend and Luma are counted only when their API responses explicitly include +valid depth, queue, and withdrawal-delay metadata. + ### GET /api/portfolio/:userId - Auth: required (requireAuth + enforceUserAccess) diff --git a/docs/LIQUIDITY_FLOOR.md b/docs/LIQUIDITY_FLOOR.md new file mode 100644 index 00000000..fc90f66a --- /dev/null +++ b/docs/LIQUIDITY_FLOOR.md @@ -0,0 +1,50 @@ +# Agent Liquidity Floor + +`User.liquidityFloor` is a minimum USD-equivalent amount the agent attempts to +keep in positions that can be exited immediately within the configured 0.5% +slippage target. It is evaluated before active goals and strategy allocation; +those controls operate only on the balance above the floor. + +## Data Contract + +Liquidity snapshots are valid for 24 hours. A position counts toward the floor +only when all of the following hold: + +- Its asset is USDC. +- Its protocol has a fresh liquidity snapshot with a valid depth curve. +- The snapshot reports zero withdrawal delay and zero queue depth. +- The position is not locked and has no pending transaction. +- The whole position is exitable within the 0.5% slippage target. + +Unknown assets, missing or stale snapshots, invalid depth curves, pending +transactions, and locked positions fail closed. The agent blocks rebalancing +while required liquidity data is unavailable rather than assuming funds are +liquid. + +The current collector derives conservative 50-basis-point depth from Horizon's +USDC pool reserves for Stellar DEX. Blend and Luma snapshots are written only +when their API responses explicitly provide withdrawal delay, queue depth, and +depth data. `ProtocolLiquiditySnapshot.withdrawalDelayHours` must be set by the +trusted collector; null means unknown, not zero. + +## Agent Behavior + +When the floor is already met, the agent allocates no more than total balance +minus the floor and never uses fully liquid positions as rebalance sources. +When there is a shortfall, it considers unlocked positions with known exit times +in shortest-exit order and routes only to protocols with verified immediate +liquidity. Existing strategy risk and rebalance cost/payback gates still apply. +Pending transactions prevent duplicate attempts. + +If the floor exceeds the total balance, the agent keeps the full balance out of +yield and reports the shortfall. A configured floor is a restoration target, not +an instantaneous on-chain guarantee; delayed withdrawals and cost gates can +extend restoration time. + +## API + +- `GET /api/v1/liquidity-floor` returns the authenticated user's configured + floor, current balances, shortfall, estimated restoration time, and data + availability. +- `PATCH /api/v1/liquidity-floor` accepts `{ "floorUsd": }`. + Send `null` to clear the floor. diff --git a/prisma/migrations/20260930200000_agent_liquidity_floor/migration.sql b/prisma/migrations/20260930200000_agent_liquidity_floor/migration.sql new file mode 100644 index 00000000..2ffb7212 --- /dev/null +++ b/prisma/migrations/20260930200000_agent_liquidity_floor/migration.sql @@ -0,0 +1,8 @@ +ALTER TABLE "users" +ADD COLUMN "liquidityFloor" DECIMAL(36, 18); + +ALTER TABLE "positions" +ADD COLUMN "liquidityLocked" BOOLEAN NOT NULL DEFAULT false; + +ALTER TABLE "protocol_liquidity_snapshots" +ADD COLUMN "withdrawalDelayHours" INTEGER; \ No newline at end of file diff --git a/prisma/migrations/20260930200000_agent_liquidity_floor/rollback.sql b/prisma/migrations/20260930200000_agent_liquidity_floor/rollback.sql new file mode 100644 index 00000000..991771e7 --- /dev/null +++ b/prisma/migrations/20260930200000_agent_liquidity_floor/rollback.sql @@ -0,0 +1,8 @@ +ALTER TABLE "protocol_liquidity_snapshots" +DROP COLUMN "withdrawalDelayHours"; + +ALTER TABLE "positions" +DROP COLUMN "liquidityLocked"; + +ALTER TABLE "users" +DROP COLUMN "liquidityFloor"; \ No newline at end of file diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 4ac56558..1f12be90 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -238,6 +238,7 @@ model User { riskTolerance Int @default(5) rebalanceStrategy String? // 'MAX_YIELD' | 'TARGET_ALLOCATION' | null (defaults to MAX_YIELD) strategyConfig Json? // e.g. { "targetAllocations": { "Blend": 50, "Stellar DEX": 30, "Luma": 20 } } + liquidityFloor Decimal? @db.Decimal(36, 18) taxAwareRebalancing Boolean @default(false) // #550 - opt-in wash-sale-aware rebalancing isActive Boolean @default(true) createdAt DateTime @default(now()) @@ -489,6 +490,7 @@ model Position { depositedAmount Decimal @db.Decimal(36, 18) currentValue Decimal @db.Decimal(36, 18) yieldEarned Decimal @default(0) @db.Decimal(36, 18) + liquidityLocked Boolean @default(false) status PositionStatus @default(ACTIVE) openedAt DateTime @default(now()) closedAt DateTime? @@ -2054,6 +2056,7 @@ model ProtocolLiquiditySnapshot { availableLiquidity Decimal @db.Decimal(36, 18) dailyVolume Decimal @db.Decimal(36, 18) withdrawalQueueDepth Int? + withdrawalDelayHours Int? depthCurve Json // [{ sizeUsd, priceImpactBps }] fetchedAt DateTime @default(now()) diff --git a/src/agent/loop.ts b/src/agent/loop.ts index 88a4abe0..d9fdc7f8 100644 --- a/src/agent/loop.ts +++ b/src/agent/loop.ts @@ -209,9 +209,13 @@ async function rebalanceCheckJob(): Promise { for (const pos of positions) { const { config, follow } = effectiveByUser.get(pos.userId)! const hasGoal = goalsByUser.has(pos.userId) + const liquidityFloorConfigured = + (pos.user as any).liquidityFloor !== null && + (pos.user as any).liquidityFloor !== undefined + const floorUserKey = liquidityFloorConfigured ? pos.userId : 'shared' const key = `${pos.protocolName}:${config.strategyName || 'DEFAULT'}:${ config.riskCeiling ?? 'none' - }:${follow?.followId ?? 'none'}:${hasGoal ? 'goal' : 'nogoal'}` + }:${follow?.followId ?? 'none'}:${hasGoal ? 'goal' : 'nogoal'}:${floorUserKey}` if (!byProtocolAndStrategy.has(key)) { byProtocolAndStrategy.set(key, { protocol: pos.protocolName, @@ -343,7 +347,10 @@ async function rebalanceCheckJob(): Promise { // null — the common case for a new user, and exactly who this feature // targets — would have had their followed config silently ignored. const hasStrategyContext = - Boolean(lead.config.strategyName) || Boolean(lead.follow) + Boolean(lead.config.strategyName) || + Boolean(lead.follow) || + (protocolPositions[0].user as any).liquidityFloor !== null && + (protocolPositions[0].user as any).liquidityFloor !== undefined const userStrategyPreferences = hasStrategyContext ? protocolPositions.map((p: PositionWithUser) => { @@ -364,6 +371,14 @@ async function rebalanceCheckJob(): Promise { // them. exposureCaps: config.exposureCaps, defaultMaxFraction: config.defaultMaxFraction, + ...((p.user as any).liquidityFloor === null || + (p.user as any).liquidityFloor === undefined + ? {} + : { + liquidityFloorUsd: Number( + (p.user as any).liquidityFloor + ), + }), followedStrategyId: follow?.followedStrategyId ?? undefined, } }) diff --git a/src/agent/router.ts b/src/agent/router.ts index 9570233b..ae4d8f37 100644 --- a/src/agent/router.ts +++ b/src/agent/router.ts @@ -35,6 +35,8 @@ import { enqueueOutboxOp } from '../outbox/service' import { dispatchInBackground } from '../outbox/dispatcher' import { deriveIdempotencyKey } from '../outbox/idempotency' import { persistRebalanceDecision } from './rebalanceDecision' +import { getLiquidityFloorStatus } from '../analytics/liquidityFloor' +import { STROOPS_PER_TOKEN } from '../config/financial-limits' const DEFAULT_THRESHOLDS: RebalanceThresholds = { minimumImprovement: 0.5, // Must improve by at least 0.5% @@ -525,9 +527,11 @@ export async function executeRebalanceIfNeeded( batchContext?: RebalanceBatchContext ): Promise { try { - const totalAmount = userPositions + let totalAmount = userPositions .reduce((sum, pos) => sum + BigInt(pos.amount), BigInt(0)) .toString() + let positionsToMove = userPositions + let liquidityRestoreTargets: Set | null = null const effectiveThresholds = thresholds ?? getThresholds() const affectedUserIds = Array.from( @@ -681,6 +685,140 @@ export async function executeRebalanceIfNeeded( return null } + const floorUsd = userStrategyPreferences[0]?.liquidityFloorUsd + if (floorUsd !== undefined && floorUsd > 0) { + const floorUserId = userStrategyPreferences[0].userId + const floorStatus = await getLiquidityFloorStatus(floorUserId) + if (!floorStatus.dataAvailable) { + await recordDecision({ + outcome: 'BLOCKED', + blockedReason: 'liquidity_data_unavailable', + rationale: + 'Liquidity floor is configured, but fresh exit and lock data is unavailable; no rebalance was made.', + trace: { + currentApy, + chosenProtocol: null, + chosenApy: null, + rawImprovement: null, + netImprovement: null, + estCostPercent: null, + costBreakdown: null, + thresholds: effectiveThresholds, + candidates: [], + }, + }) + return null + } + + if (floorStatus.shortfallUsd > 0) { + const targetCapacities = Object.entries( + floorStatus.instantProtocolCapacityUsd + ).filter(([protocol]) => protocol !== currentProtocol) + const restoreLimit = Math.min( + floorStatus.shortfallUsd, + Math.max(0, ...targetCapacities.map(([, capacity]) => capacity)) + ) + const plannedByPosition = new Map() + let remainingRestore = restoreLimit + for (const item of floorStatus.restorationPlan) { + if (remainingRestore <= 0) break + const amount = Math.min(item.amountUsd, remainingRestore) + plannedByPosition.set(item.positionId, amount) + remainingRestore -= amount + } + positionsToMove = userPositions.filter((position) => + plannedByPosition.has(position.id) + ) + const restoreUsd = Array.from(plannedByPosition.values()).reduce( + (sum, amount) => sum + amount, + 0 + ) + liquidityRestoreTargets = new Set( + targetCapacities + .filter(([, capacity]) => capacity >= restoreUsd) + .map(([protocol]) => protocol) + ) + + if (positionsToMove.length === 0 || liquidityRestoreTargets.size === 0) { + await recordDecision({ + outcome: 'BLOCKED', + blockedReason: 'liquidity_floor_shortfall', + rationale: + 'The liquidity floor is below target, but no unlocked exit source and verified liquid target are available.', + trace: { + currentApy, + chosenProtocol: null, + chosenApy: null, + rawImprovement: null, + netImprovement: null, + estCostPercent: null, + costBreakdown: null, + thresholds: effectiveThresholds, + candidates: [], + }, + }) + return null + } + + const restoreAmount = BigInt( + Math.floor(restoreUsd * STROOPS_PER_TOKEN) + ) + const positionAmount = positionsToMove.reduce( + (sum, position) => sum + BigInt(position.amount), + BigInt(0) + ) + totalAmount = (restoreAmount < positionAmount + ? restoreAmount + : positionAmount + ).toString() + allProtocols = allProtocols.filter((protocol) => + liquidityRestoreTargets!.has(protocol.name) + ) + } else { + const instantLiquidIds = new Set( + floorStatus.instantLiquidPositionIds + ) + const lockedIds = new Set(floorStatus.lockedPositionIds) + positionsToMove = userPositions.filter( + (position) => + !instantLiquidIds.has(position.id) && !lockedIds.has(position.id) + ) + const yieldAmount = BigInt( + Math.floor(floorStatus.availableForYieldUsd * STROOPS_PER_TOKEN) + ) + const portfolioAmount = BigInt(totalAmount) + const movableAmount = positionsToMove.reduce( + (sum, position) => sum + BigInt(position.amount), + BigInt(0) + ) + const yieldLimit = + yieldAmount < portfolioAmount ? yieldAmount : portfolioAmount + totalAmount = (yieldLimit < movableAmount + ? yieldLimit + : movableAmount + ).toString() + if (totalAmount === '0' || positionsToMove.length === 0) { + await recordDecision({ + outcome: 'HELD', + rationale: + 'The configured liquidity floor leaves no balance available for yield allocation.', + trace: { + currentApy, + chosenProtocol: null, + chosenApy: null, + rawImprovement: null, + netImprovement: null, + estCostPercent: null, + costBreakdown: null, + thresholds: effectiveThresholds, + candidates: [], + }, + }) + return null + } + } + } + // An ACTIVE savings goal (#281) takes priority over the stored strategy // preference — a user working toward a stated target/date should have // the agent chase whatever rate that goal actually needs, not a static @@ -697,7 +835,9 @@ export async function executeRebalanceIfNeeded( } const preferredStrategy = userStrategyPreferences[0]?.strategyName - const strategy: RebalanceStrategy = activeGoal + const strategy: RebalanceStrategy = liquidityRestoreTargets + ? new MaxYieldStrategy() + : activeGoal ? new GoalTrackingStrategy() : preferredStrategy === 'TARGET_ALLOCATION' ? new TargetAllocationStrategy() @@ -827,7 +967,13 @@ export async function executeRebalanceIfNeeded( // If the preferred target itself had zero headroom, still try a real // move into the next allocation; otherwise nothing can move. const moves = plan.allocations - .filter((a) => a.fraction > 0 && a.protocol !== currentProtocol) + .filter( + (a) => + a.fraction > 0 && + a.protocol !== currentProtocol && + (!liquidityRestoreTargets || + liquidityRestoreTargets.has(a.protocol)) + ) .map((a) => ({ toProtocol: a.protocol, fraction: a.fraction, @@ -884,7 +1030,7 @@ export async function executeRebalanceIfNeeded( currentProtocol, first.toProtocol, amountToMove.toString(), - userPositions.map((pos) => pos.id), + positionsToMove.map((pos) => pos.id), { name: strategy.name, reasoning: decision.reasoning, diff --git a/src/agent/scanner.ts b/src/agent/scanner.ts index e57c7f09..4685f9b0 100644 --- a/src/agent/scanner.ts +++ b/src/agent/scanner.ts @@ -10,6 +10,11 @@ import { effectiveApy, shouldUseEffectiveApy, } from '../analytics/yieldComposition' +import { + buildExplicitProtocolLiquiditySnapshot, + buildStellarDexLiquiditySnapshot, + recordLiquiditySnapshotSafely, +} from '../analytics/liquiditySnapshot' const ASSET_SYMBOL = 'USDC' const MINIMUM_TVL = 10000 @@ -69,6 +74,15 @@ async function fetchBlendApy(): Promise { if (apyRate === null) throw new Error('Could not parse Blend APY from response') + const liquiditySnapshot = buildExplicitProtocolLiquiditySnapshot( + 'Blend', + ASSET_SYMBOL, + reserve + ) + if (liquiditySnapshot) { + await recordLiquiditySnapshotSafely(liquiditySnapshot) + } + recordMetric('Blend', Date.now() - start, false) return { @@ -106,6 +120,13 @@ async function fetchStellarDexApy(): Promise { const pools = data?._embedded?.records || [] if (pools.length === 0) throw new Error('No Stellar DEX pools found') + const liquiditySnapshot = buildStellarDexLiquiditySnapshot( + pools, + `${ASSET_SYMBOL}:${usdcIssuer}` + ) + if (liquiditySnapshot) { + await recordLiquiditySnapshotSafely(liquiditySnapshot) + } // Aggregate: weighted average fee APY by TVL let totalTvl = 0 @@ -161,6 +182,15 @@ async function fetchLumaApy(): Promise { const apyRate = parseFloat(rate.apy) * 100 const tvl = rate.tvl ? parseFloat(rate.tvl) : undefined + const liquiditySnapshot = buildExplicitProtocolLiquiditySnapshot( + 'Luma', + ASSET_SYMBOL, + rate + ) + if (liquiditySnapshot) { + await recordLiquiditySnapshotSafely(liquiditySnapshot) + } + recordMetric('Luma', Date.now() - start, false) return { diff --git a/src/agent/tools/actionTools.ts b/src/agent/tools/actionTools.ts index 50ec801b..7ddafe7a 100644 --- a/src/agent/tools/actionTools.ts +++ b/src/agent/tools/actionTools.ts @@ -209,6 +209,7 @@ export const rebalanceTool = defineTool({ rebalanceStrategy: true, strategyConfig: true, riskTolerance: true, + liquidityFloor: true, }, }) if (!user) return { ok: false, error: 'User not found' } @@ -243,7 +244,9 @@ export const rebalanceTool = defineTool({ const thresholds = getThresholds() const hasStrategyContext = - Boolean(effective.strategyName) || Boolean(follow) + Boolean(effective.strategyName) || + Boolean(follow) || + user.liquidityFloor !== null const preferences = hasStrategyContext ? [ { @@ -252,6 +255,9 @@ export const rebalanceTool = defineTool({ targetAllocations: effective.targetAllocations, riskTolerance: user.riskTolerance, riskCeiling: effective.riskCeiling, + ...(user.liquidityFloor === null + ? {} + : { liquidityFloorUsd: Number(user.liquidityFloor) }), followedStrategyId: follow?.followedStrategyId ?? undefined, }, ] diff --git a/src/agent/types.ts b/src/agent/types.ts index 3590f865..3296825a 100644 --- a/src/agent/types.ts +++ b/src/agent/types.ts @@ -246,6 +246,8 @@ export interface UserStrategyPreferences { * table but yields to a per-protocol override. */ defaultMaxFraction?: number + liquidityFloorUsd?: number + taxAwareRebalancing?: boolean } /** diff --git a/src/analytics/liquidity.ts b/src/analytics/liquidity.ts index 8b6e07ff..2d2ead40 100644 --- a/src/analytics/liquidity.ts +++ b/src/analytics/liquidity.ts @@ -27,6 +27,83 @@ export interface LiquidityMetrics { bindingConstraint?: 'depth' | 'cooldown' | 'none' } +export interface LiquidityFloorPosition { + positionId: string + valueUsd: number + timeToExitHours: number | null + locked: boolean +} + +export interface LiquidityFloorRestoration { + positionId: string + amountUsd: number + timeToExitHours: number +} + +export function calculateLiquidityFloor(params: { + totalBalanceUsd: number + liquidBalanceUsd: number + floorUsd: number +}): { + availableForYieldUsd: number + shortfallUsd: number +} { + const { totalBalanceUsd, liquidBalanceUsd, floorUsd } = params + if ( + !Number.isFinite(totalBalanceUsd) || + !Number.isFinite(liquidBalanceUsd) || + !Number.isFinite(floorUsd) || + totalBalanceUsd < 0 || + liquidBalanceUsd < 0 || + floorUsd < 0 + ) { + return { availableForYieldUsd: 0, shortfallUsd: 0 } + } + + return { + availableForYieldUsd: Math.max(0, totalBalanceUsd - floorUsd), + shortfallUsd: Math.max(0, floorUsd - liquidBalanceUsd), + } +} + +export function planLiquidityFloorRestoration( + positions: LiquidityFloorPosition[], + shortfallUsd: number +): LiquidityFloorRestoration[] { + if (!Number.isFinite(shortfallUsd) || shortfallUsd <= 0) return [] + + const candidates = positions + .filter( + (position) => + !position.locked && + Number.isFinite(position.valueUsd) && + position.valueUsd > 0 && + position.timeToExitHours !== null && + Number.isFinite(position.timeToExitHours) && + position.timeToExitHours >= 0 + ) + .sort( + (a, b) => + a.timeToExitHours! - b.timeToExitHours! || + a.positionId.localeCompare(b.positionId) + ) + + let remaining = shortfallUsd + const restoration: LiquidityFloorRestoration[] = [] + for (const position of candidates) { + if (remaining <= 0) break + const amountUsd = Math.min(position.valueUsd, remaining) + restoration.push({ + positionId: position.positionId, + amountUsd, + timeToExitHours: position.timeToExitHours!, + }) + remaining -= amountUsd + } + + return restoration +} + // ── Pure Core Functions ──────────────────────────────────────────────────────── export function maxExitWithinSlippage( diff --git a/src/analytics/liquidityFloor.ts b/src/analytics/liquidityFloor.ts new file mode 100644 index 00000000..3f9de016 --- /dev/null +++ b/src/analytics/liquidityFloor.ts @@ -0,0 +1,243 @@ +import db from '../db' +import { STROOPS_PER_TOKEN } from '../config/financial-limits' +import { + calculateLiquidityFloor, + LIQUIDITY_CONFIG, + maxExitWithinSlippage, + planLiquidityFloorRestoration, + timeToFullExit, + type DepthCurvePoint, + type LiquidityFloorPosition, + type LiquidityFloorRestoration, +} from './liquidity' + +const RESERVE_ASSET = 'USDC' + +export interface LiquidityFloorStatus { + floorUsd: number | null + totalBalanceUsd: number + liquidBalanceUsd: number + availableForYieldUsd: number + shortfallUsd: number + estimatedRestoreHours: number | null + dataAvailable: boolean + floorExceedsBalance: boolean + restorationPlan: LiquidityFloorRestoration[] + instantProtocolNames: string[] + instantProtocolCapacityUsd: Record + instantLiquidPositionIds: string[] + lockedPositionIds: string[] +} + +function parseDepthCurve(value: unknown): DepthCurvePoint[] | null { + if (!Array.isArray(value)) return null + const points: DepthCurvePoint[] = [] + for (const item of value) { + if ( + typeof item !== 'object' || + item === null || + !('sizeUsd' in item) || + !('priceImpactBps' in item) + ) { + return null + } + const point = item as Record + if ( + typeof point.sizeUsd !== 'number' || + !Number.isFinite(point.sizeUsd) || + point.sizeUsd < 0 || + typeof point.priceImpactBps !== 'number' || + !Number.isFinite(point.priceImpactBps) || + point.priceImpactBps < 0 + ) { + return null + } + points.push({ + sizeUsd: point.sizeUsd, + priceImpactBps: point.priceImpactBps, + }) + } + return points +} + +export async function getLiquidityFloorStatus( + userId: string, + now = new Date() +): Promise { + const user = await db.user.findUnique({ + where: { id: userId }, + select: { liquidityFloor: true }, + }) + if (!user) throw new Error('User not found') + + const positions = await db.position.findMany({ + where: { userId, status: 'ACTIVE' }, + select: { + id: true, + protocolName: true, + assetSymbol: true, + currentValue: true, + liquidityLocked: true, + transactions: { + where: { status: 'PENDING' }, + select: { id: true }, + }, + }, + }) + + const reservePositions = positions.filter( + (position) => position.assetSymbol.toUpperCase() === RESERVE_ASSET + ) + const unsupportedAssetsPresent = positions.length !== reservePositions.length + const cutoff = new Date(now.getTime() - LIQUIDITY_CONFIG.SNAPSHOT_MAX_AGE_MS) + const snapshots = await db.protocolLiquiditySnapshot.findMany({ + where: { + assetSymbol: RESERVE_ASSET, + fetchedAt: { gte: cutoff }, + }, + orderBy: { fetchedAt: 'desc' }, + }) + + const latestByProtocol = new Map() + for (const snapshot of snapshots) { + if (!latestByProtocol.has(snapshot.protocolName)) { + latestByProtocol.set(snapshot.protocolName, snapshot) + } + } + const instantProtocolCapacityUsd: Record = {} + for (const [protocolName, snapshot] of latestByProtocol.entries()) { + const curve = parseDepthCurve(snapshot.depthCurve) + const capacity = + curve === null || !Number.isFinite(Number(snapshot.availableLiquidity)) + ? 0 + : maxExitWithinSlippage( + curve, + Number(snapshot.availableLiquidity), + LIQUIDITY_CONFIG.TARGET_SLIPPAGE_BPS + ) + if ( + curve !== null && + curve.length > 0 && + capacity > 0 && + snapshot.withdrawalQueueDepth === 0 && + snapshot.withdrawalDelayHours === 0 + ) { + instantProtocolCapacityUsd[protocolName] = capacity + } + } + const instantProtocolNames = Object.keys(instantProtocolCapacityUsd) + + let totalBalanceUsd = 0 + let liquidBalanceUsd = 0 + let dataAvailable = !unsupportedAssetsPresent + const restorationPositions: LiquidityFloorPosition[] = [] + const instantLiquidPositionIds: string[] = [] + const lockedPositionIds: string[] = [] + + for (const position of reservePositions) { + if (position.liquidityLocked) lockedPositionIds.push(position.id) + if ((position.transactions?.length ?? 0) > 0) { + dataAvailable = false + continue + } + const valueUsd = Number(position.currentValue) / STROOPS_PER_TOKEN + if (!Number.isFinite(valueUsd) || valueUsd < 0) { + dataAvailable = false + continue + } + totalBalanceUsd += valueUsd + + const snapshot = latestByProtocol.get(position.protocolName) + const depthCurve = snapshot ? parseDepthCurve(snapshot.depthCurve) : null + if ( + !snapshot || + depthCurve === null || + !Number.isFinite(Number(snapshot.availableLiquidity)) || + Number(snapshot.availableLiquidity) <= 0 || + snapshot.withdrawalQueueDepth === null || + snapshot.withdrawalDelayHours === null + ) { + dataAvailable = false + continue + } + + const maxExit = maxExitWithinSlippage( + depthCurve, + Number(snapshot.availableLiquidity), + LIQUIDITY_CONFIG.TARGET_SLIPPAGE_BPS + ) + const instantlyLiquid = + !position.liquidityLocked && + snapshot.withdrawalQueueDepth === 0 && + snapshot.withdrawalDelayHours === 0 + + if (instantlyLiquid && maxExit >= valueUsd) { + liquidBalanceUsd += valueUsd + instantLiquidPositionIds.push(position.id) + continue + } + + const timeToExit = + !position.liquidityLocked && + maxExit > 0 && + snapshot.withdrawalDelayHours > 0 + ? timeToFullExit({ + positionValue: valueUsd, + maxExitPerSlice: maxExit, + sliceIntervalHours: + snapshot.withdrawalDelayHours * + Math.max(1, snapshot.withdrawalQueueDepth + 1), + depthRecoveryModel: 'linear', + }) + : null + restorationPositions.push({ + positionId: position.id, + valueUsd: + snapshot.withdrawalDelayHours === 0 + ? Math.min(valueUsd, maxExit) + : valueUsd, + timeToExitHours: + snapshot.withdrawalDelayHours === 0 ? 0 : (timeToExit?.hours ?? null), + locked: position.liquidityLocked, + }) + } + + const floorUsd = + user.liquidityFloor === null ? null : Number(user.liquidityFloor) + const floor = floorUsd ?? 0 + const floorPlan = calculateLiquidityFloor({ + totalBalanceUsd, + liquidBalanceUsd, + floorUsd: floor, + }) + const restorationPlan = planLiquidityFloorRestoration( + restorationPositions, + floorPlan.shortfallUsd + ) + const plannedRestoreUsd = restorationPlan.reduce( + (sum, item) => sum + item.amountUsd, + 0 + ) + const estimatedRestoreHours = + floorPlan.shortfallUsd === 0 + ? 0 + : plannedRestoreUsd >= floorPlan.shortfallUsd + ? restorationPlan.reduce((sum, item) => sum + item.timeToExitHours, 0) + : null + + return { + floorUsd, + totalBalanceUsd, + liquidBalanceUsd, + availableForYieldUsd: dataAvailable ? floorPlan.availableForYieldUsd : 0, + shortfallUsd: floorPlan.shortfallUsd, + estimatedRestoreHours, + dataAvailable, + floorExceedsBalance: dataAvailable && floor > totalBalanceUsd, + restorationPlan, + instantProtocolNames, + instantProtocolCapacityUsd, + instantLiquidPositionIds, + lockedPositionIds, + } +} diff --git a/src/analytics/liquiditySnapshot.ts b/src/analytics/liquiditySnapshot.ts new file mode 100644 index 00000000..ad10e9b3 --- /dev/null +++ b/src/analytics/liquiditySnapshot.ts @@ -0,0 +1,168 @@ +import { Prisma } from '@prisma/client' +import db from '../db' +import { logger } from '../utils/logger' +import { LIQUIDITY_CONFIG, type DepthCurvePoint } from './liquidity' + +export interface ProtocolLiquiditySnapshotInput { + protocolName: string + assetSymbol: string + poolTvl: number + availableLiquidity: number + dailyVolume: number + withdrawalQueueDepth: number | null + withdrawalDelayHours: number | null + depthCurve: DepthCurvePoint[] +} + +function parseProviderNumber(value: unknown): number | null { + if (typeof value === 'number') { + return Number.isFinite(value) ? value : null + } + if (typeof value === 'string' && value.trim() !== '') { + const parsed = Number(value) + return Number.isFinite(parsed) ? parsed : null + } + return null +} + +export function buildExplicitProtocolLiquiditySnapshot( + protocolName: string, + assetSymbol: string, + source: unknown +): ProtocolLiquiditySnapshotInput | null { + if (typeof source !== 'object' || source === null) return null + const data = source as Record + const availableLiquidity = parseProviderNumber(data.availableLiquidity) + const withdrawalQueueDepth = parseProviderNumber(data.withdrawalQueueDepth) + const withdrawalDelayHours = parseProviderNumber(data.withdrawalDelayHours) + if ( + availableLiquidity === null || + availableLiquidity <= 0 || + withdrawalQueueDepth === null || + !Number.isInteger(withdrawalQueueDepth) || + withdrawalQueueDepth < 0 || + withdrawalDelayHours === null || + !Number.isInteger(withdrawalDelayHours) || + withdrawalDelayHours < 0 || + !Array.isArray(data.depthCurve) + ) { + return null + } + + const depthCurve: DepthCurvePoint[] = [] + for (const point of data.depthCurve) { + if (typeof point !== 'object' || point === null) return null + const sizeUsd = parseProviderNumber( + (point as Record).sizeUsd + ) + const priceImpactBps = parseProviderNumber( + (point as Record).priceImpactBps + ) + if ( + sizeUsd === null || + sizeUsd < 0 || + priceImpactBps === null || + priceImpactBps < 0 + ) { + return null + } + depthCurve.push({ sizeUsd, priceImpactBps }) + } + if (depthCurve.length === 0) return null + + const poolTvl = parseProviderNumber(data.poolTvl ?? data.totalSupply) + const dailyVolume = parseProviderNumber(data.dailyVolume ?? 0) + return { + protocolName, + assetSymbol, + poolTvl: poolTvl !== null && poolTvl >= 0 ? poolTvl : availableLiquidity, + availableLiquidity, + dailyVolume: dailyVolume !== null && dailyVolume >= 0 ? dailyVolume : 0, + withdrawalQueueDepth, + withdrawalDelayHours, + depthCurve, + } +} + +export function buildStellarDexLiquiditySnapshot( + pools: unknown[], + usdcAsset: string +): ProtocolLiquiditySnapshotInput | null { + let availableLiquidity = 0 + for (const pool of pools) { + if (typeof pool !== 'object' || pool === null || !('reserves' in pool)) { + continue + } + const reserves = (pool as { reserves?: unknown }).reserves + if (!Array.isArray(reserves)) continue + const reserve = reserves.find( + (item) => + typeof item === 'object' && + item !== null && + 'asset' in item && + (item as { asset?: unknown }).asset === usdcAsset + ) as { amount?: unknown } | undefined + const amount = Number(reserve?.amount) + if (Number.isFinite(amount) && amount > 0) availableLiquidity += amount + } + + if (availableLiquidity <= 0) return null + const maxExitAtTarget = + (availableLiquidity * LIQUIDITY_CONFIG.TARGET_SLIPPAGE_BPS) / + (10_000 - LIQUIDITY_CONFIG.TARGET_SLIPPAGE_BPS) + + return { + protocolName: 'Stellar DEX', + assetSymbol: 'USDC', + poolTvl: availableLiquidity, + availableLiquidity, + dailyVolume: 0, + withdrawalQueueDepth: 0, + withdrawalDelayHours: 0, + depthCurve: [ + { + sizeUsd: maxExitAtTarget, + priceImpactBps: LIQUIDITY_CONFIG.TARGET_SLIPPAGE_BPS, + }, + ], + } +} + +export async function recordLiquiditySnapshot( + snapshot: ProtocolLiquiditySnapshotInput, + now = new Date() +): Promise { + const recent = await db.protocolLiquiditySnapshot.findFirst({ + where: { + protocolName: snapshot.protocolName, + assetSymbol: snapshot.assetSymbol, + fetchedAt: { + gte: new Date(now.getTime() - LIQUIDITY_CONFIG.SNAPSHOT_MAX_AGE_MS), + }, + }, + select: { id: true }, + }) + if (recent) return + + await db.protocolLiquiditySnapshot.create({ + data: { + ...snapshot, + depthCurve: snapshot.depthCurve as unknown as Prisma.InputJsonValue, + fetchedAt: now, + }, + }) +} + +export async function recordLiquiditySnapshotSafely( + snapshot: ProtocolLiquiditySnapshotInput +): Promise { + try { + await recordLiquiditySnapshot(snapshot) + } catch (error) { + logger.warn('Could not persist protocol liquidity snapshot', { + protocolName: snapshot.protocolName, + assetSymbol: snapshot.assetSymbol, + error: error instanceof Error ? error.message : 'Unknown error', + }) + } +} diff --git a/src/index.ts b/src/index.ts index 477af739..4b6dfb5b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -98,6 +98,7 @@ import keysRouter from './routes/keys' import sessionsRouter from './routes/sessions' import streamRouter from './routes/stream' import notificationsRouter from './routes/notifications' +import liquidityFloorRouter from './routes/liquidity-floor' import networkRouter from './routes/network' import { corsMiddleware, @@ -355,6 +356,7 @@ const apiRoutes: ApiRoute[] = [ { path: 'sessions', handlers: [sessionsRouter] }, { path: 'stream', handlers: [streamRouter] }, { path: 'notifications', handlers: [notificationsRouter] }, + { path: 'liquidity-floor', handlers: [liquidityFloorRouter] }, { path: 'admin', handlers: [adminRateLimiter, adminRouter] }, ] diff --git a/src/routes/liquidity-floor.ts b/src/routes/liquidity-floor.ts new file mode 100644 index 00000000..99a08dc3 --- /dev/null +++ b/src/routes/liquidity-floor.ts @@ -0,0 +1,63 @@ +import { Router, Request, Response } from 'express' +import { z } from 'zod' +import db from '../db' +import { requireAuth } from '../middleware/authenticate' +import { validate } from '../middleware/validate' +import { getLiquidityFloorStatus } from '../analytics/liquidityFloor' +import { sendError } from '../utils/errors' + +const router = Router() + +const updateLiquidityFloorSchema = z.object({ + body: z.object({ + floorUsd: z.number().finite().min(0).nullable(), + }), +}) + +router.use(requireAuth) + +router.get('/', async (req: Request, res: Response) => { + const userId = req.userId + if (!userId) return sendError(res, 401, 'Unauthorized') + + try { + const status = await getLiquidityFloorStatus(userId) + return res.json({ + floorUsd: status.floorUsd, + totalBalanceUsd: status.totalBalanceUsd, + liquidBalanceUsd: status.liquidBalanceUsd, + availableForYieldUsd: status.availableForYieldUsd, + shortfallUsd: status.shortfallUsd, + estimatedRestoreHours: status.estimatedRestoreHours, + dataAvailable: status.dataAvailable, + floorExceedsBalance: status.floorExceedsBalance, + }) + } catch (error) { + if (error instanceof Error && error.message === 'User not found') { + return sendError(res, 404, 'User not found') + } + throw error + } +}) + +router.patch( + '/', + validate({ body: updateLiquidityFloorSchema.shape.body }), + async (req: Request, res: Response) => { + const userId = req.userId + if (!userId) return sendError(res, 401, 'Unauthorized') + + const { floorUsd } = req.body as z.infer< + typeof updateLiquidityFloorSchema.shape.body + > + const user = await db.user.updateMany({ + where: { id: userId }, + data: { liquidityFloor: floorUsd === null ? null : String(floorUsd) }, + }) + if (user.count === 0) return sendError(res, 404, 'User not found') + + return res.json(await getLiquidityFloorStatus(userId)) + } +) + +export default router diff --git a/tests/integration/agent/rebalanceCost.integration.test.ts b/tests/integration/agent/rebalanceCost.integration.test.ts index 19ffa554..760a3e9f 100644 --- a/tests/integration/agent/rebalanceCost.integration.test.ts +++ b/tests/integration/agent/rebalanceCost.integration.test.ts @@ -37,11 +37,26 @@ const mockTransactionCreate = jest .mockImplementation(({ data }: any) => ({ id: 'txn-1', ...data })) const mockPositionFindFirst = jest.fn().mockResolvedValue(null) const mockPositionFindMany = jest.fn().mockResolvedValue([]) +const mockUserFindUnique = jest.fn() +const mockSnapshotFindMany = jest.fn() +const mockSavingsGoalFindFirst = jest.fn() const mockOutboxOpCreate = jest.fn().mockResolvedValue({ id: 'op-1' }) const mockOutboxOpFindUnique = jest.fn().mockResolvedValue(null) +const mockPersistDecision = jest.fn().mockResolvedValue('decision-1') + +jest.mock('../../../src/agent/rebalanceDecision', () => ({ + persistRebalanceDecision: (...args: unknown[]) => mockPersistDecision(...args), +})) jest.mock('../../../src/db', () => { const client: any = { + user: { findUnique: (...a: unknown[]) => mockUserFindUnique(...a) }, + savingsGoal: { + findFirst: (...a: unknown[]) => mockSavingsGoalFindFirst(...a), + }, + protocolLiquiditySnapshot: { + findMany: (...a: unknown[]) => mockSnapshotFindMany(...a), + }, agentLog: { create: (...a: unknown[]) => mockAgentLogCreate(...a) }, transaction: { create: (...a: unknown[]) => mockTransactionCreate(...a) }, position: { @@ -84,6 +99,10 @@ describe('#347 rebalance cost payback gate (integration)', () => { assetSymbol: 'USDC', user: { network: 'MAINNET' }, }) + mockUserFindUnique.mockResolvedValue({ liquidityFloor: '100' }) + mockSnapshotFindMany.mockResolvedValue([]) + mockSavingsGoalFindFirst.mockResolvedValue(null) + mockPersistDecision.mockResolvedValue('decision-1') }) it('blocks an unprofitable move: tiny position whose cost never recoups', async () => { @@ -111,4 +130,33 @@ describe('#347 rebalance cost payback gate (integration)', () => { expect(mockTransactionCreate).toHaveBeenCalledTimes(1) expect(mockOutboxOpCreate).toHaveBeenCalledTimes(1) }) + + it('blocks a configured floor when fresh liquidity data is unavailable', async () => { + mockPositionFindMany.mockResolvedValue([ + { + id: 'pos-big', + protocolName: 'Blend', + assetSymbol: 'USDC', + currentValue: '10000000000000000000000', + liquidityLocked: false, + transactions: [], + }, + ]) + + const result = await executeRebalanceIfNeeded( + 'Blend', + [{ id: 'pos-big', amount: '10000000000000000000000', userId: 'user-1' }], + undefined, + [{ userId: 'user-1', liquidityFloorUsd: 100 }] + ) + + expect(result).toBeNull() + expect(mockSubmitRebalance).not.toHaveBeenCalled() + expect(mockPersistDecision).toHaveBeenCalledWith( + expect.objectContaining({ + outcome: 'BLOCKED', + blockedReason: 'liquidity_data_unavailable', + }) + ) + }) }) diff --git a/tests/unit/analytics/liquidity.test.ts b/tests/unit/analytics/liquidity.test.ts new file mode 100644 index 00000000..189e2fa3 --- /dev/null +++ b/tests/unit/analytics/liquidity.test.ts @@ -0,0 +1,80 @@ +import { + calculateLiquidityFloor, + planLiquidityFloorRestoration, +} from '../../../src/analytics/liquidity' + +describe('calculateLiquidityFloor', () => { + it('limits yield allocation to total balance minus the minimum floor', () => { + expect( + calculateLiquidityFloor({ + totalBalanceUsd: 1000, + liquidBalanceUsd: 700, + floorUsd: 500, + }) + ).toEqual({ availableForYieldUsd: 500, shortfallUsd: 0 }) + }) + + it('reports a shortfall and keeps all funds liquid when floor exceeds balance', () => { + expect( + calculateLiquidityFloor({ + totalBalanceUsd: 100, + liquidBalanceUsd: 0, + floorUsd: 500, + }) + ).toEqual({ availableForYieldUsd: 0, shortfallUsd: 500 }) + }) + + it('fails closed on invalid inputs', () => { + expect( + calculateLiquidityFloor({ + totalBalanceUsd: Number.NaN, + liquidBalanceUsd: 0, + floorUsd: 10, + }) + ).toEqual({ availableForYieldUsd: 0, shortfallUsd: 0 }) + }) +}) + +describe('planLiquidityFloorRestoration', () => { + it('restores from the shortest exits first and skips locked or unknown positions', () => { + expect( + planLiquidityFloorRestoration( + [ + { + positionId: 'slow', + valueUsd: 500, + timeToExitHours: 12, + locked: false, + }, + { + positionId: 'fast', + valueUsd: 100, + timeToExitHours: 1, + locked: false, + }, + { + positionId: 'locked', + valueUsd: 500, + timeToExitHours: 0, + locked: true, + }, + { + positionId: 'unknown', + valueUsd: 500, + timeToExitHours: null, + locked: false, + }, + ], + 250 + ) + ).toEqual([ + { positionId: 'fast', amountUsd: 100, timeToExitHours: 1 }, + { positionId: 'slow', amountUsd: 150, timeToExitHours: 12 }, + ]) + }) + + it('does not create a restoration plan for an invalid or empty shortfall', () => { + expect(planLiquidityFloorRestoration([], 0)).toEqual([]) + expect(planLiquidityFloorRestoration([], Number.NaN)).toEqual([]) + }) +}) diff --git a/tests/unit/analytics/liquidityFloorService.test.ts b/tests/unit/analytics/liquidityFloorService.test.ts new file mode 100644 index 00000000..afc869f0 --- /dev/null +++ b/tests/unit/analytics/liquidityFloorService.test.ts @@ -0,0 +1,248 @@ +const mockDb = { + user: { findUnique: jest.fn() }, + position: { findMany: jest.fn() }, + protocolLiquiditySnapshot: { + findFirst: jest.fn(), + findMany: jest.fn(), + create: jest.fn(), + }, +} + +jest.mock('../../../src/db', () => ({ __esModule: true, default: mockDb })) + +import db from '../../../src/db' +import { getLiquidityFloorStatus } from '../../../src/analytics/liquidityFloor' +import { + buildExplicitProtocolLiquiditySnapshot, + buildStellarDexLiquiditySnapshot, + recordLiquiditySnapshot, +} from '../../../src/analytics/liquiditySnapshot' + +const mockedDb = db as any +const NOW = new Date('2026-09-30T12:00:00.000Z') + +function resetDb() { + jest.resetAllMocks() + mockedDb.user.findUnique.mockResolvedValue({ liquidityFloor: '50' }) + mockedDb.position.findMany.mockResolvedValue([ + { + id: 'position-1', + protocolName: 'Stellar DEX', + assetSymbol: 'USDC', + currentValue: '1000000000', + liquidityLocked: false, + }, + ]) + mockedDb.protocolLiquiditySnapshot.findMany.mockResolvedValue([ + { + protocolName: 'Stellar DEX', + assetSymbol: 'USDC', + availableLiquidity: '1000', + withdrawalQueueDepth: 0, + withdrawalDelayHours: 0, + depthCurve: [{ sizeUsd: 1000, priceImpactBps: 0 }], + fetchedAt: NOW, + }, + ]) + mockedDb.protocolLiquiditySnapshot.findFirst.mockResolvedValue(null) +} + +describe('buildStellarDexLiquiditySnapshot', () => { + it('aggregates matching USDC reserves as immediately withdrawable liquidity', () => { + expect( + buildStellarDexLiquiditySnapshot( + [ + { + reserves: [ + { asset: 'USDC:ISSUER', amount: '120' }, + { asset: 'XLM:native', amount: '40' }, + ], + }, + { reserves: [{ asset: 'USDC:ISSUER', amount: '30' }] }, + ], + 'USDC:ISSUER' + ) + ).toEqual({ + protocolName: 'Stellar DEX', + assetSymbol: 'USDC', + poolTvl: 150, + availableLiquidity: 150, + dailyVolume: 0, + withdrawalQueueDepth: 0, + withdrawalDelayHours: 0, + depthCurve: [ + { + sizeUsd: (150 * 50) / 9950, + priceImpactBps: 50, + }, + ], + }) + }) + + it('does not classify missing or invalid reserve data as liquid', () => { + expect( + buildStellarDexLiquiditySnapshot( + [{ reserves: [{ asset: 'USDC:ISSUER', amount: 'NaN' }] }], + 'USDC:ISSUER' + ) + ).toBeNull() + }) +}) + +describe('buildExplicitProtocolLiquiditySnapshot', () => { + it('accepts provider data only when delay, queue, and depth are explicit', () => { + expect( + buildExplicitProtocolLiquiditySnapshot('Blend', 'USDC', { + availableLiquidity: 1000, + withdrawalQueueDepth: 0, + withdrawalDelayHours: 0, + depthCurve: [{ sizeUsd: 100, priceImpactBps: 50 }], + }) + ).toMatchObject({ + protocolName: 'Blend', + assetSymbol: 'USDC', + availableLiquidity: 1000, + withdrawalQueueDepth: 0, + withdrawalDelayHours: 0, + }) + }) + + it('rejects provider data that omits withdrawal delay', () => { + expect( + buildExplicitProtocolLiquiditySnapshot('Luma', 'USDC', { + availableLiquidity: 1000, + withdrawalQueueDepth: 0, + depthCurve: [{ sizeUsd: 100, priceImpactBps: 50 }], + }) + ).toBeNull() + }) + + it('does not coerce a null withdrawal delay to zero', () => { + expect( + buildExplicitProtocolLiquiditySnapshot('Blend', 'USDC', { + availableLiquidity: 1000, + withdrawalQueueDepth: 0, + withdrawalDelayHours: null, + depthCurve: [{ sizeUsd: 100, priceImpactBps: 50 }], + }) + ).toBeNull() + }) +}) + +describe('liquidity snapshot persistence', () => { + beforeEach(resetDb) + + it('avoids inserting another fresh snapshot for the same protocol and asset', async () => { + mockedDb.protocolLiquiditySnapshot.findFirst.mockResolvedValue({ + id: 'existing', + }) + await recordLiquiditySnapshot( + { + protocolName: 'Stellar DEX', + assetSymbol: 'USDC', + poolTvl: 100, + availableLiquidity: 100, + dailyVolume: 0, + withdrawalQueueDepth: 0, + withdrawalDelayHours: 0, + depthCurve: [{ sizeUsd: 100, priceImpactBps: 0 }], + }, + NOW + ) + expect(mockedDb.protocolLiquiditySnapshot.create).not.toHaveBeenCalled() + }) +}) + +describe('getLiquidityFloorStatus', () => { + beforeEach(resetDb) + + it('counts only a fresh, fully exitable, unlocked USDC position', async () => { + const status = await getLiquidityFloorStatus('user-1', NOW) + expect(status).toMatchObject({ + floorUsd: 50, + totalBalanceUsd: 100, + liquidBalanceUsd: 100, + availableForYieldUsd: 50, + shortfallUsd: 0, + dataAvailable: true, + instantLiquidPositionIds: ['position-1'], + }) + }) + + it('excludes locked positions and reports their floor shortfall', async () => { + mockedDb.position.findMany.mockResolvedValue([ + { + id: 'position-1', + protocolName: 'Stellar DEX', + assetSymbol: 'USDC', + currentValue: '1000000000', + liquidityLocked: true, + }, + ]) + const status = await getLiquidityFloorStatus('user-1', NOW) + expect(status).toMatchObject({ + liquidBalanceUsd: 0, + shortfallUsd: 50, + estimatedRestoreHours: null, + dataAvailable: true, + instantLiquidPositionIds: [], + }) + }) + + it('fails closed when a fresh snapshot is missing', async () => { + mockedDb.protocolLiquiditySnapshot.findMany.mockResolvedValue([]) + const status = await getLiquidityFloorStatus('user-1', NOW) + expect(status).toMatchObject({ + liquidBalanceUsd: 0, + shortfallUsd: 50, + dataAvailable: false, + availableForYieldUsd: 0, + instantProtocolNames: [], + }) + }) + + it('does not report a complete restoration time when known exits cannot cover the shortfall', async () => { + mockedDb.user.findUnique.mockResolvedValue({ liquidityFloor: '500' }) + mockedDb.position.findMany.mockResolvedValue([ + { + id: 'position-1', + protocolName: 'Stellar DEX', + assetSymbol: 'USDC', + currentValue: '1000000000', + liquidityLocked: false, + transactions: [], + }, + ]) + mockedDb.protocolLiquiditySnapshot.findMany.mockResolvedValue([ + { + protocolName: 'Stellar DEX', + assetSymbol: 'USDC', + availableLiquidity: '100', + withdrawalQueueDepth: 0, + withdrawalDelayHours: 0, + depthCurve: [{ sizeUsd: 1, priceImpactBps: 50 }], + fetchedAt: NOW, + }, + ]) + const status = await getLiquidityFloorStatus('user-1', NOW) + expect(status.estimatedRestoreHours).toBeNull() + }) + + it('fails closed while a position transaction is pending', async () => { + mockedDb.position.findMany.mockResolvedValue([ + { + id: 'position-1', + protocolName: 'Stellar DEX', + assetSymbol: 'USDC', + currentValue: '1000000000', + liquidityLocked: false, + transactions: [{ id: 'pending-withdrawal' }], + }, + ]) + const status = await getLiquidityFloorStatus('user-1', NOW) + expect(status).toMatchObject({ + liquidBalanceUsd: 0, + dataAvailable: false, + }) + }) +}) diff --git a/tests/unit/routes/liquidity-floor.test.ts b/tests/unit/routes/liquidity-floor.test.ts new file mode 100644 index 00000000..f1afd6a8 --- /dev/null +++ b/tests/unit/routes/liquidity-floor.test.ts @@ -0,0 +1,102 @@ +process.env.NODE_ENV = 'test' + +import express, { NextFunction, Request, Response } from 'express' +import request from 'supertest' +import { Network } from '@prisma/client' +import liquidityFloorRouter from '../../../src/routes/liquidity-floor' + +const mockUserFindUnique = jest.fn() +const mockUserUpdateMany = jest.fn() +const mockPositionFindMany = jest.fn() +const mockSnapshotFindMany = jest.fn() + +jest.mock('../../../src/db', () => ({ + __esModule: true, + default: { + user: { + findUnique: (...args: unknown[]) => mockUserFindUnique(...args), + updateMany: (...args: unknown[]) => mockUserUpdateMany(...args), + }, + position: { + findMany: (...args: unknown[]) => mockPositionFindMany(...args), + }, + protocolLiquiditySnapshot: { + findMany: (...args: unknown[]) => mockSnapshotFindMany(...args), + }, + }, +})) + +jest.mock('../../../src/middleware/authenticate', () => ({ + requireAuth: (req: Request, res: Response, next: NextFunction) => { + if (!req.headers.authorization) { + res.status(401).json({ error: 'Unauthorized' }) + return + } + req.userId = 'caller-id' + req.auth = { + userId: 'caller-id', + sessionId: 'session-id', + walletAddress: 'GTEST', + network: Network.MAINNET, + } + next() + }, +})) + +const app = express() +app.use(express.json()) +app.use('/liquidity-floor', liquidityFloorRouter) + +beforeEach(() => { + jest.clearAllMocks() + mockUserFindUnique.mockResolvedValue({ liquidityFloor: '50' }) + mockUserUpdateMany.mockResolvedValue({ count: 1 }) + mockPositionFindMany.mockResolvedValue([]) + mockSnapshotFindMany.mockResolvedValue([]) +}) + +describe('liquidity-floor routes', () => { + it('requires authentication', async () => { + const response = await request(app).get('/liquidity-floor') + expect(response.status).toBe(401) + }) + + it('updates only the authenticated user and returns status', async () => { + mockUserFindUnique.mockResolvedValue({ liquidityFloor: '250' }) + const response = await request(app) + .patch('/liquidity-floor') + .set('Authorization', 'Bearer test-token') + .send({ floorUsd: 250 }) + + expect(response.status).toBe(200) + expect(mockUserUpdateMany).toHaveBeenCalledWith({ + where: { id: 'caller-id' }, + data: { liquidityFloor: '250' }, + }) + expect(response.body.floorUsd).toBe(250) + }) + + it('clears the floor when floorUsd is null', async () => { + mockUserFindUnique.mockResolvedValue({ liquidityFloor: null }) + const response = await request(app) + .patch('/liquidity-floor') + .set('Authorization', 'Bearer test-token') + .send({ floorUsd: null }) + + expect(response.status).toBe(200) + expect(mockUserUpdateMany).toHaveBeenCalledWith({ + where: { id: 'caller-id' }, + data: { liquidityFloor: null }, + }) + }) + + it('rejects negative floors', async () => { + const response = await request(app) + .patch('/liquidity-floor') + .set('Authorization', 'Bearer test-token') + .send({ floorUsd: -1 }) + + expect(response.status).toBe(400) + expect(mockUserUpdateMany).not.toHaveBeenCalled() + }) +})