From c3e2ee141e0a3b8a74f7065e83652e4a86804179 Mon Sep 17 00:00:00 2001 From: gospeltout Date: Sat, 17 Oct 2026 12:12:57 +0100 Subject: [PATCH] feat: ingest Blend lending pool state, rates, and derived utilisation --- .env.example | 22 + openapi.json | 257 +++++++ openapi.yaml | 164 +++++ .../migration.sql | 30 + prisma/schema.prisma | 25 + src/__tests__/blendIngester.test.ts | 169 +++++ src/__tests__/blendUtilization.test.ts | 200 ++++++ src/__tests__/envDrift.test.ts | 4 + src/__tests__/lendingRateRetention.test.ts | 111 +++ src/__tests__/ratesRoute.test.ts | 131 ++++ src/app.ts | 2 + src/config.ts | 59 ++ src/db.ts | 74 ++ src/index.ts | 36 +- src/ingest/venues/blend.ts | 632 ++++++++++++++++++ src/ingest/venues/index.ts | 1 + src/jobs/lendingRateRetention.ts | 131 ++++ src/metrics.ts | 14 + src/routes/rates.ts | 108 +++ 19 files changed, 2169 insertions(+), 1 deletion(-) create mode 100644 prisma/migrations/20261017000000_create_lending_rates/migration.sql create mode 100644 src/__tests__/blendIngester.test.ts create mode 100644 src/__tests__/blendUtilization.test.ts create mode 100644 src/__tests__/lendingRateRetention.test.ts create mode 100644 src/__tests__/ratesRoute.test.ts create mode 100644 src/ingest/venues/blend.ts create mode 100644 src/jobs/lendingRateRetention.ts create mode 100644 src/routes/rates.ts diff --git a/.env.example b/.env.example index 41e4465f..41e8fe6e 100644 --- a/.env.example +++ b/.env.example @@ -60,6 +60,12 @@ AQUARIUS_ENABLED_TESTNET=false # Reflector oracle contract (56-char C... id) — see https://reflector.network. Empty = oracle disabled. # A malformed id disables the oracle on this network and logs a warning naming this variable. REFLECTOR_CONTRACT_ID_TESTNET= +# Blend lending pools (comma-separated 56-char C... ids) on testnet. +BLEND_POOLS_TESTNET=CCEBVDYM32YNYCVNRXQKDFFPISJJCV557CDZEIRBEE4NCV4KHPQ44HGF +# Blend backstop contract on testnet (56-char C... id). +BLEND_BACKSTOP_TESTNET=CBDVWXT433PRVTUNM56C3JREF3HIZHRBA64NB2C3B2UNCKIS65ZYCLZA +# Blend lending rate ingestion enabled on testnet (default: true). +BLEND_ENABLED_TESTNET=true # Comma-separated pairs to watch on testnet. # Format: "CODE:ISSUER/CODE:ISSUER". Use "native" for XLM. WATCHED_PAIRS_TESTNET=XLM:native/USDC:GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5 @@ -77,6 +83,12 @@ AQUARIUS_ENABLED_MAINNET=true # Reflector oracle contract on mainnet (56-char C... id) — copy it from https://reflector.network. # There is no built-in default: leave empty to keep the oracle disabled. REFLECTOR_CONTRACT_ID_MAINNET= +# Blend lending pools (comma-separated 56-char C... ids) on mainnet. +BLEND_POOLS_MAINNET=CAJJZSGMMM3PD7N33TAPHGBUGTB43OC73HVIK2L2G6BNGGGYOSSYBXBD,CCCCIQSDILITHMM7PBSLVDT5MISSY7R26MNZXCX4H7J5JQ5FPIYOGYFS,CDVQVKOY2YSXS2IC7KN6MNASSHPAO7UN2UR2ON4OI2SKMFJNVAMDX6DP,CBP7NO6F7FRDHSOFQBT2L2UWYIZ2PU76JKVRYAQTG3KZSQLYAOKIF2WB +# Blend backstop contract on mainnet (56-char C... id). +BLEND_BACKSTOP_MAINNET=CAQQR5SWBXKIGZKPBZDH3KM5GQ5GUTPKB7JAFCINLZBC5WXPJKRG3IM7 +# Blend lending rate ingestion enabled on mainnet (default: true). +BLEND_ENABLED_MAINNET=true # Comma-separated pairs to watch on mainnet. WATCHED_PAIRS_MAINNET=XLM:native/USDC:GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN @@ -97,6 +109,12 @@ SOROSWAP_FACTORY_ADDRESS=CDP3HMUH6SMS3S7NPGNDJLULCOXXEPSHY4JKUKMBNQMATHDHWXRRJTB WATCHED_PAIRS=XLM:native/USDC:GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5 # Reflector oracle contract ID on testnet (falls back from paired REFLECTOR_CONTRACT_ID_TESTNET) REFLECTOR_CONTRACT_ID= +# Blend lending pools on testnet (falls back from paired BLEND_POOLS_TESTNET) +BLEND_POOLS=CCEBVDYM32YNYCVNRXQKDFFPISJJCV557CDZEIRBEE4NCV4KHPQ44HGF +# Blend backstop contract on testnet (falls back from paired BLEND_BACKSTOP_TESTNET) +BLEND_BACKSTOP=CBDVWXT433PRVTUNM56C3JREF3HIZHRBA64NB2C3B2UNCKIS65ZYCLZA +# Blend lending rate ingestion enabled (falls back from paired BLEND_ENABLED_TESTNET) +BLEND_ENABLED=true # --- Indexer Configuration --- # Frequency of polling for new trades/snapshots in milliseconds (default: 5000) @@ -107,6 +125,8 @@ SDEX_PAGE_SIZE=200 AMM_PAGE_SIZE=200 # How often to poll Soroswap pool reserves in milliseconds (default: 60000 = 60s) SOROSWAP_POLL_INTERVAL_MS=60000 +# How often to poll Blend pool reserves in milliseconds (default: 60000 = 60s) +BLEND_POLL_INTERVAL_MS=60000 # --- Auth & Security --- # Secret key required for admin endpoints (e.g., adding/removing pairs) @@ -191,3 +211,5 @@ FACILITATOR_FEE_STROOPS_MAINNET=50000 # against a table that has never been pruned cannot hold a long lock or blow # the WAL. Default 1000. SNAPSHOT_PRUNE_BATCH_SIZE=1000 +# Rows removed by one DELETE during a lending rate retention run (default 1000). +RATE_PRUNE_BATCH_SIZE=1000 diff --git a/openapi.json b/openapi.json index cb6441b7..f6f6d273 100644 --- a/openapi.json +++ b/openapi.json @@ -1858,6 +1858,144 @@ } } } + }, + "/rates": { + "get": { + "operationId": "getLendingRates", + "summary": "Query latest lending and borrow rates across pools", + "description": "Returns the latest supply and borrow rates, utilisation, and reserves for lending markets.", + "security": [], + "parameters": [ + { + "name": "asset", + "in": "query", + "required": false, + "description": "Filter by asset symbol (e.g. USDC, XLM)", + "schema": { + "type": "string" + } + }, + { + "name": "poolId", + "in": "query", + "required": false, + "description": "Filter by pool contract ID", + "schema": { + "type": "string" + } + }, + { + "name": "network", + "in": "query", + "required": false, + "description": "Stellar network to query (testnet or mainnet)", + "schema": { + "type": "string", + "enum": [ + "testnet", + "mainnet" + ] + } + }, + { + "name": "limit", + "in": "query", + "required": false, + "description": "Maximum number of rate records to return", + "schema": { + "type": "integer", + "default": 100 + } + } + ], + "responses": { + "200": { + "description": "Latest lending rates", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/LendingRatesResponse" + } + } + } + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "429": { + "$ref": "#/components/responses/TooManyRequests" + } + } + } + }, + "/rates/{asset}": { + "get": { + "operationId": "getLendingRatesForAsset", + "summary": "Query latest lending and borrow rates for a specific asset", + "description": "Returns the latest supply and borrow rates, utilisation, and reserves for a specific asset (e.g. USDC, XLM).", + "security": [], + "parameters": [ + { + "name": "asset", + "in": "path", + "required": true, + "description": "Asset symbol (e.g. USDC, XLM)", + "schema": { + "type": "string" + } + }, + { + "name": "poolId", + "in": "query", + "required": false, + "description": "Filter by pool contract ID", + "schema": { + "type": "string" + } + }, + { + "name": "network", + "in": "query", + "required": false, + "description": "Stellar network to query (testnet or mainnet)", + "schema": { + "type": "string", + "enum": [ + "testnet", + "mainnet" + ] + } + }, + { + "name": "limit", + "in": "query", + "required": false, + "description": "Maximum number of rate records to return", + "schema": { + "type": "integer", + "default": 100 + } + } + ], + "responses": { + "200": { + "description": "Latest lending rates for the requested asset", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/AssetLendingRatesResponse" + } + } + } + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "429": { + "$ref": "#/components/responses/TooManyRequests" + } + } + } } }, "components": { @@ -1977,6 +2115,125 @@ } }, "schemas": { + "LendingRateEntry": { + "type": "object", + "required": [ + "id", + "venue", + "source", + "network", + "poolId", + "asset", + "supplyApy", + "borrowApy", + "utilization", + "timestamp" + ], + "properties": { + "id": { + "type": "string" + }, + "venue": { + "type": "string", + "example": "blend" + }, + "source": { + "type": "string", + "example": "blend" + }, + "network": { + "type": "string", + "example": "mainnet" + }, + "poolId": { + "type": "string" + }, + "asset": { + "type": "string", + "example": "USDC" + }, + "assetAddress": { + "type": "string", + "nullable": true + }, + "supplyApy": { + "type": "number" + }, + "borrowApy": { + "type": "number" + }, + "supplyApr": { + "type": "number", + "nullable": true + }, + "borrowApr": { + "type": "number", + "nullable": true + }, + "utilization": { + "type": "number" + }, + "totalSupply": { + "type": "number", + "nullable": true + }, + "totalBorrowed": { + "type": "number", + "nullable": true + }, + "reserves": { + "type": "number", + "nullable": true + }, + "ledger": { + "type": "integer" + }, + "timestamp": { + "type": "string", + "format": "date-time" + } + } + }, + "LendingRatesResponse": { + "type": "object", + "required": [ + "network", + "rates" + ], + "properties": { + "network": { + "type": "string" + }, + "rates": { + "type": "array", + "items": { + "$ref": "#/components/schemas/LendingRateEntry" + } + } + } + }, + "AssetLendingRatesResponse": { + "type": "object", + "required": [ + "asset", + "network", + "rates" + ], + "properties": { + "asset": { + "type": "string" + }, + "network": { + "type": "string" + }, + "rates": { + "type": "array", + "items": { + "$ref": "#/components/schemas/LendingRateEntry" + } + } + } + }, "Asset": { "type": "object", "properties": { diff --git a/openapi.yaml b/openapi.yaml index 9d511d0b..bbee52ca 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -1197,6 +1197,96 @@ paths: '402': $ref: '#/components/responses/PaymentRequired' + /rates: + get: + operationId: getLendingRates + summary: Query latest lending and borrow rates across pools + description: Returns the latest supply and borrow rates, utilisation, and reserves for lending markets. + security: [] + parameters: + - name: asset + in: query + required: false + description: Filter by asset symbol (e.g. USDC, XLM) + schema: + type: string + - name: poolId + in: query + required: false + description: Filter by pool contract ID + schema: + type: string + - name: network + in: query + required: false + description: Stellar network to query (testnet or mainnet) + schema: + type: string + enum: [testnet, mainnet] + - name: limit + in: query + required: false + description: Maximum number of rate records to return + schema: + type: integer + default: 100 + responses: + '200': + description: Latest lending rates + content: + application/json: + schema: + $ref: '#/components/schemas/LendingRatesResponse' + '400': + $ref: '#/components/responses/BadRequest' + '429': + $ref: '#/components/responses/TooManyRequests' + + /rates/{asset}: + get: + operationId: getLendingRatesForAsset + summary: Query latest lending and borrow rates for a specific asset + description: Returns the latest supply and borrow rates, utilisation, and reserves for a specific asset (e.g. USDC, XLM). + security: [] + parameters: + - name: asset + in: path + required: true + description: Asset symbol (e.g. USDC, XLM) + schema: + type: string + - name: poolId + in: query + required: false + description: Filter by pool contract ID + schema: + type: string + - name: network + in: query + required: false + description: Stellar network to query (testnet or mainnet) + schema: + type: string + enum: [testnet, mainnet] + - name: limit + in: query + required: false + description: Maximum number of rate records to return + schema: + type: integer + default: 100 + responses: + '200': + description: Latest lending rates for the requested asset + content: + application/json: + schema: + $ref: '#/components/schemas/AssetLendingRatesResponse' + '400': + $ref: '#/components/responses/BadRequest' + '429': + $ref: '#/components/responses/TooManyRequests' + components: securitySchemes: x402: @@ -1283,6 +1373,80 @@ components: type: string schemas: + LendingRateEntry: + type: object + required: [id, venue, source, network, poolId, asset, supplyApy, borrowApy, utilization, timestamp] + properties: + id: + type: string + venue: + type: string + example: blend + source: + type: string + example: blend + network: + type: string + example: mainnet + poolId: + type: string + asset: + type: string + example: USDC + assetAddress: + type: string + nullable: true + supplyApy: + type: number + borrowApy: + type: number + supplyApr: + type: number + nullable: true + borrowApr: + type: number + nullable: true + utilization: + type: number + totalSupply: + type: number + nullable: true + totalBorrowed: + type: number + nullable: true + reserves: + type: number + nullable: true + ledger: + type: integer + timestamp: + type: string + format: date-time + + LendingRatesResponse: + type: object + required: [network, rates] + properties: + network: + type: string + rates: + type: array + items: + $ref: '#/components/schemas/LendingRateEntry' + + AssetLendingRatesResponse: + type: object + required: [asset, network, rates] + properties: + asset: + type: string + network: + type: string + rates: + type: array + items: + $ref: '#/components/schemas/LendingRateEntry' + Asset: type: object properties: diff --git a/prisma/migrations/20261017000000_create_lending_rates/migration.sql b/prisma/migrations/20261017000000_create_lending_rates/migration.sql new file mode 100644 index 00000000..c2bd249f --- /dev/null +++ b/prisma/migrations/20261017000000_create_lending_rates/migration.sql @@ -0,0 +1,30 @@ +-- CreateTable +CREATE TABLE "lending_rates" ( + "id" TEXT NOT NULL, + "network" TEXT NOT NULL DEFAULT 'testnet', + "source" TEXT NOT NULL DEFAULT 'blend', + "pool_id" TEXT NOT NULL, + "asset" TEXT NOT NULL, + "asset_address" TEXT, + "supply_apy" DECIMAL(36,18) NOT NULL, + "borrow_apy" DECIMAL(36,18) NOT NULL, + "supply_apr" DECIMAL(36,18), + "borrow_apr" DECIMAL(36,18), + "utilization" DECIMAL(36,18) NOT NULL, + "total_supply" DECIMAL(36,7), + "total_borrowed" DECIMAL(36,7), + "reserves" DECIMAL(36,7), + "ledger" INTEGER NOT NULL DEFAULT 0, + "timestamp" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "lending_rates_pkey" PRIMARY KEY ("id","timestamp") +); + +-- CreateIndex +CREATE INDEX "lending_rates_network_pool_id_timestamp_idx" ON "lending_rates"("network", "pool_id", "timestamp" DESC); + +-- CreateIndex +CREATE INDEX "lending_rates_network_asset_timestamp_idx" ON "lending_rates"("network", "asset", "timestamp" DESC); + +-- CreateIndex +CREATE INDEX "lending_rates_network_pool_id_asset_timestamp_idx" ON "lending_rates"("network", "pool_id", "asset", "timestamp" DESC); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 29189728..adc401df 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -270,3 +270,28 @@ model ApiKey { @@map("api_keys") } + +model LendingRate { + id String @default(uuid()) + network String @default("testnet") + source String @default("blend") + poolId String @map("pool_id") + asset String + assetAddress String? @map("asset_address") + supplyApy Decimal @map("supply_apy") @db.Decimal(36, 18) + borrowApy Decimal @map("borrow_apy") @db.Decimal(36, 18) + supplyApr Decimal? @map("supply_apr") @db.Decimal(36, 18) + borrowApr Decimal? @map("borrow_apr") @db.Decimal(36, 18) + utilization Decimal @db.Decimal(36, 18) + totalSupply Decimal? @map("total_supply") @db.Decimal(36, 7) + totalBorrowed Decimal? @map("total_borrowed") @db.Decimal(36, 7) + reserves Decimal? @db.Decimal(36, 7) + ledger Int @default(0) + timestamp DateTime + + @@id([id, timestamp]) + @@index([network, poolId, timestamp(sort: Desc)]) + @@index([network, asset, timestamp(sort: Desc)]) + @@index([network, poolId, asset, timestamp(sort: Desc)]) + @@map("lending_rates") +} diff --git a/src/__tests__/blendIngester.test.ts b/src/__tests__/blendIngester.test.ts new file mode 100644 index 00000000..96986463 --- /dev/null +++ b/src/__tests__/blendIngester.test.ts @@ -0,0 +1,169 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import { + ingestBlendPool, + ingestBlendPools, + blendAdapter, + _resetLastPrice, + _getLastPrice, + type BlendPoolData, +} from '../ingest/venues/blend' +import * as dbModule from '../db' +import { blend_pool_errors_total, blend_rates_ingested_total } from '../metrics' + +vi.mock('../db', () => ({ + insertLendingRates: vi.fn().mockResolvedValue(1), + getLatestLendingRates: vi.fn().mockResolvedValue([]), +})) + +describe('Blend Ingester & Fault Isolation', () => { + beforeEach(() => { + vi.clearAllMocks() + _resetLastPrice() + }) + + const mockPoolId = 'CAJJZSGMMM3PD7N33TAPHGBUGTB43OC73HVIK2L2G6BNGGGYOSSYBXBD' + const mockAssetId = 'CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75' // USDC + + const samplePoolData: BlendPoolData = { + poolId: mockPoolId, + network: 'mainnet', + backstopRate: 1_000_000n, // 10% + reserves: [ + { + assetId: mockAssetId, + assetSymbol: 'USDC', + rateDecimals: 12, + config: { + index: 0, + decimals: 7, + util: 7_500_000, + r_base: 100_000, + r_one: 400_000, + r_two: 2_000_000, + r_three: 5_000_000, + }, + data: { + b_supply: 2_000_000_0000000n, // 2M USDC + d_supply: 1_000_000_0000000n, // 1M USDC borrowed => 50% util + b_rate: 10n ** 12n, + d_rate: 10n ** 12n, + ir_mod: 10_000_000n, + }, + }, + ], + } + + it('successfully ingests a healthy pool and writes rates with network', async () => { + const mockFetch = vi.fn().mockResolvedValue(samplePoolData) + const incSpy = vi.spyOn(blend_rates_ingested_total, 'inc') + + const rates = await ingestBlendPool(mockPoolId, 'mainnet', mockFetch) + + expect(rates.length).toBe(1) + expect(rates[0].poolId).toBe(mockPoolId) + expect(rates[0].asset).toBe('USDC') + expect(rates[0].assetAddress).toBe(mockAssetId) + expect(rates[0].source).toBe('blend') + expect(rates[0].utilization).toBe(0.5) + expect(rates[0].totalSupply).toBe(2_000_000) + expect(rates[0].totalBorrowed).toBe(1_000_000) + expect(rates[0].reserves).toBe(1_000_000) + + // Must write to database with explicit network + expect(dbModule.insertLendingRates).toHaveBeenCalledTimes(1) + expect(dbModule.insertLendingRates).toHaveBeenCalledWith( + expect.arrayContaining([ + expect.objectContaining({ + poolId: mockPoolId, + asset: 'USDC', + source: 'blend', + }), + ]), + 'mainnet' + ) + + expect(incSpy).toHaveBeenCalledWith({ pool: mockPoolId, network: 'mainnet' }, 1) + }) + + it('skips a pool that returns null/empty with a metric and log, never crashing', async () => { + const mockFetch = vi.fn().mockResolvedValue(null) + const errMetricSpy = vi.spyOn(blend_pool_errors_total, 'inc') + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}) + + // Must not throw + const rates = await ingestBlendPool(mockPoolId, 'testnet', mockFetch) + + expect(rates).toEqual([]) + expect(dbModule.insertLendingRates).not.toHaveBeenCalled() + expect(errMetricSpy).toHaveBeenCalledWith({ pool: mockPoolId, network: 'testnet' }) + expect(warnSpy).toHaveBeenCalled() + }) + + it('skips a pool that throws an RPC network error with a metric and log, never crashing', async () => { + const mockFetch = vi.fn().mockRejectedValue(new Error('RPC request timed out')) + const errMetricSpy = vi.spyOn(blend_pool_errors_total, 'inc') + const errLogSpy = vi.spyOn(console, 'error').mockImplementation(() => {}) + + // Must not throw or crash + const rates = await ingestBlendPool(mockPoolId, 'mainnet', mockFetch) + + expect(rates).toEqual([]) + expect(dbModule.insertLendingRates).not.toHaveBeenCalled() + expect(errMetricSpy).toHaveBeenCalledWith({ pool: mockPoolId, network: 'mainnet' }) + expect(errLogSpy).toHaveBeenCalledWith( + expect.stringContaining('[blend] Failed to ingest pool'), + 'RPC request timed out' + ) + }) + + it('continues ingesting other pools when one pool fails in ingestBlendPools', async () => { + const pool1 = 'CAJJZSGMMM3PD7N33TAPHGBUGTB43OC73HVIK2L2G6BNGGGYOSSYBXBD' + const pool2 = 'CCCCIQSDILITHMM7PBSLVDT5MISSY7R26MNZXCX4H7J5JQ5FPIYOGYFS' + + const mockIngest = vi.fn().mockImplementation(async (poolId) => { + if (poolId === pool1) { + throw new Error('Simulated pool corruption') + } + return [ + { + poolId: pool2, + asset: 'USDC', + supplyApy: 0.05, + borrowApy: 0.08, + utilization: 0.6, + }, + ] + }) + + const count = await ingestBlendPools('mainnet', mockIngest) + // Should still ingest the second pool + expect(count).toBeGreaterThanOrEqual(1) + expect(mockIngest).toHaveBeenCalled() + }) + + it('implements VenueAdapter contract and reference helper functions', () => { + expect(blendAdapter.name).toBe('Blend') + expect(blendAdapter.source).toBe('blend') + expect(typeof blendAdapter.isEnabled).toBe('function') + expect(typeof blendAdapter.getPollIntervalMs).toBe('function') + expect(typeof blendAdapter.discover).toBe('function') + expect(typeof blendAdapter.read).toBe('function') + expect(typeof blendAdapter.derive).toBe('function') + expect(typeof blendAdapter.write).toBe('function') + expect(typeof blendAdapter.ingestPair).toBe('function') + + // Reset last price helper + _resetLastPrice() + expect(_getLastPrice('mainnet', 'XLM/USDC')).toBeUndefined() + }) + + it('enforces network isolation: testnet and mainnet never cross-contaminate', async () => { + const mockFetch = vi.fn().mockResolvedValue(samplePoolData) + + await ingestBlendPool(mockPoolId, 'testnet', mockFetch) + expect(dbModule.insertLendingRates).toHaveBeenLastCalledWith(expect.any(Array), 'testnet') + + await ingestBlendPool(mockPoolId, 'mainnet', mockFetch) + expect(dbModule.insertLendingRates).toHaveBeenLastCalledWith(expect.any(Array), 'mainnet') + }) +}) diff --git a/src/__tests__/blendUtilization.test.ts b/src/__tests__/blendUtilization.test.ts new file mode 100644 index 00000000..1ab02964 --- /dev/null +++ b/src/__tests__/blendUtilization.test.ts @@ -0,0 +1,200 @@ +import { describe, it, expect } from 'vitest' +import { + calcUtilization, + deriveUtilization, + calcInterestRates, + deriveReserveRates, + type BlendReserveEntry, + type BlendReserveConfig, + type BlendReserveData, +} from '../ingest/venues/blend' + +describe('Blend Utilisation Arithmetic (#pinning)', () => { + describe('calcUtilization', () => { + it('returns exact ratio for typical utilization', () => { + expect(calcUtilization(500, 1000)).toBe(0.5) + expect(calcUtilization(750, 1000)).toBe(0.75) + expect(calcUtilization(10, 100)).toBe(0.1) + expect(calcUtilization(1, 3)).toBeCloseTo(1 / 3, 6) + }) + + it('returns 0 when total liabilities are 0', () => { + expect(calcUtilization(0, 1000)).toBe(0) + expect(calcUtilization(0n, 1000n)).toBe(0) + }) + + it('returns 0 when total supply is 0 or negative', () => { + expect(calcUtilization(500, 0)).toBe(0) + expect(calcUtilization(500n, 0n)).toBe(0) + expect(calcUtilization(500, -100)).toBe(0) + expect(calcUtilization(500n, -100n)).toBe(0) + }) + + it('returns 0 when liabilities are negative', () => { + expect(calcUtilization(-50, 100)).toBe(0) + expect(calcUtilization(-50n, 100n)).toBe(0) + }) + + it('returns 1.0 when liabilities equal supply (100% utilization)', () => { + expect(calcUtilization(1000, 1000)).toBe(1.0) + expect(calcUtilization(1000n, 1000n)).toBe(1.0) + }) + + it('caps utilization at 1.0 when liabilities exceed supply', () => { + expect(calcUtilization(1200, 1000)).toBe(1.0) + expect(calcUtilization(2000n, 1000n)).toBe(1.0) + }) + + it('handles BigInt amounts accurately without precision loss', () => { + // 5 million USDC (7 decimals) + const liabilities = 5_000_000_0000000n + const supply = 10_000_000_0000000n + expect(calcUtilization(liabilities, supply)).toBe(0.5) + }) + + it('handles non-finite numbers safely', () => { + expect(calcUtilization(NaN, 1000)).toBe(0) + expect(calcUtilization(100, NaN)).toBe(0) + expect(calcUtilization(Infinity, 1000)).toBe(0) + expect(calcUtilization(100, Infinity)).toBe(0) + }) + }) + + describe('deriveUtilization', () => { + it('derives utilization from dTokens and bTokens with 12 rateDecimals (Blend v2)', () => { + const rateDecimals = 12 + const unitRate = 10n ** 12n // 1.0 rate + const dSupply = 600_0000000n + const bSupply = 1000_0000000n + + const util = deriveUtilization(dSupply, unitRate, bSupply, unitRate, rateDecimals) + expect(util).toBe(0.6) + }) + + it('properly applies ceil to liabilities and floor to supply', () => { + const rateDecimals = 12 + const scalar = 10n ** 12n + // dSupply * dRate produces a remainder => ceil should add 1 + const dSupply = 100n + const dRate = scalar + 1n // > 1.0 + // bSupply * bRate produces a remainder => floor discards it + const bSupply = 200n + const bRate = scalar + 1n + + const util = deriveUtilization(dSupply, dRate, bSupply, bRate, rateDecimals) + expect(util).toBeGreaterThan(0.5) + expect(util).toBeLessThan(0.51) + }) + }) + + describe('calcInterestRates', () => { + const standardConfig: BlendReserveConfig = { + index: 0, + decimals: 7, + util: 7_500_000, // 75% target util + r_base: 100_000, // 1% base rate + r_one: 400_000, // 4% slope 1 + r_two: 2_000_000, // 20% slope 2 + r_three: 5_000_000, // 50% slope 3 + c_factor: 7_500_000, + l_factor: 8_000_000, + } + + const standardData: BlendReserveData = { + ir_mod: 10_000_000n, // 1.0 modifier + } + + it('calculates Tier 1 rates when utilization <= targetUtil', () => { + // At 0% utilization: borrowApr should be r_base (1%) + const at0 = calcInterestRates(standardConfig, standardData, 0) + expect(at0.borrowApr).toBeCloseTo(0.01, 4) + expect(at0.supplyApr).toBe(0) // supplyCapture is 0 at 0 util + + // At target util (75%): borrowApr should be r_base + r_one = 1% + 4% = 5% + const at75 = calcInterestRates(standardConfig, standardData, 0.75) + expect(at75.borrowApr).toBeCloseTo(0.05, 4) + // supplyCapture = (1 - 0.10) * 0.75 = 0.675 + // supplyApr = 0.05 * 0.675 = 0.03375 (3.375%) + expect(at75.supplyApr).toBeCloseTo(0.03375, 4) + }) + + it('calculates Tier 2 rates when targetUtil < utilization <= 95%', () => { + // Halfway between 75% and 95% is 85% + // utilScalar = (0.85 - 0.75) / (0.95 - 0.75) = 0.10 / 0.20 = 0.5 + // baseRate = 0.5 * 20% + 4% + 1% = 10% + 5% = 15% + const at85 = calcInterestRates(standardConfig, standardData, 0.85) + expect(at85.borrowApr).toBeCloseTo(0.15, 3) + // supplyCapture = (1 - 0.10) * 0.85 = 0.765 + // supplyApr = 0.15 * 0.765 = 0.11475 + expect(at85.supplyApr).toBeCloseTo(0.11475, 3) + }) + + it('calculates Tier 3 rates when utilization > 95%', () => { + // At 100% utilization: + // utilScalar = (1.0 - 0.95) / 0.05 = 1.0 + // extraRate = 1.0 * 50% = 50% + // intersection = 1.0 * (20% + 4% + 1%) = 25% + // borrowApr = 50% + 25% = 75% + const at100 = calcInterestRates(standardConfig, standardData, 1.0) + expect(at100.borrowApr).toBeCloseTo(0.75, 3) + }) + + it('computes compound APY correctly from APR', () => { + const rates = calcInterestRates(standardConfig, standardData, 0.75) + // borrowApy: daily compounding (365 days) + const expectedBorrowApy = Math.pow(1 + rates.borrowApr / 365, 365) - 1 + expect(rates.borrowApy).toBeCloseTo(expectedBorrowApy, 6) + + // supplyApy: weekly compounding (52 weeks) + const expectedSupplyApy = Math.pow(1 + rates.supplyApr / 52, 52) - 1 + expect(rates.supplyApy).toBeCloseTo(expectedSupplyApy, 6) + + // Compound APY must always be >= nominal APR for positive rates + expect(rates.borrowApy).toBeGreaterThanOrEqual(rates.borrowApr) + expect(rates.supplyApy).toBeGreaterThanOrEqual(rates.supplyApr) + }) + + it('scales with interestRateModifier', () => { + const highModifierData: BlendReserveData = { + ir_mod: 20_000_000n, // 2.0x modifier + } + const at75 = calcInterestRates(standardConfig, highModifierData, 0.75) + // At 75% util: baseRate is 5%, multiplied by 2.0 = 10% + expect(at75.borrowApr).toBeCloseTo(0.10, 4) + }) + }) + + describe('deriveReserveRates', () => { + it('correctly scales token decimals to float amounts', () => { + const entry: BlendReserveEntry = { + assetId: 'CAS3J7GYLGXMF6TDJBBYYSE3HQ6BBSMLNUQ34T6TZMYMW2EZH3MS2UY7', + assetSymbol: 'XLM', + rateDecimals: 12, + config: { + index: 0, + decimals: 7, + util: 7_500_000, + r_base: 100_000, + r_one: 400_000, + r_two: 2_000_000, + r_three: 5_000_000, + }, + data: { + b_supply: 1_000_000_0000000n, // 1M XLM supplied (7 decimals) + d_supply: 400_000_0000000n, // 400k XLM borrowed + b_rate: 10n ** 12n, + d_rate: 10n ** 12n, + ir_mod: 10_000_000n, + }, + } + + const derived = deriveReserveRates(entry, 1_000_000n) + expect(derived.totalSupply).toBe(1_000_000) + expect(derived.totalBorrowed).toBe(400_000) + expect(derived.reserves).toBe(600_000) + expect(derived.utilization).toBe(0.4) + expect(derived.borrowApy).toBeGreaterThan(0) + expect(derived.supplyApy).toBeGreaterThan(0) + }) + }) +}) diff --git a/src/__tests__/envDrift.test.ts b/src/__tests__/envDrift.test.ts index 2187798b..8d7ce017 100644 --- a/src/__tests__/envDrift.test.ts +++ b/src/__tests__/envDrift.test.ts @@ -38,6 +38,10 @@ export const UNDOCUMENTED_ENV_ALLOW_LIST: Record = { 'Optional per-network feature flag override; enabled by default on testnet if reflectorContractId is configured.', REFLECTOR_ENABLED_MAINNET: 'Optional per-network feature flag override; enabled by default on mainnet if reflectorContractId is configured.', + BLEND_POLL_INTERVAL_MS_TESTNET: + 'Optional per-network override for polling interval; shared fallback BLEND_POLL_INTERVAL_MS is documented in .env.example.', + BLEND_POLL_INTERVAL_MS_MAINNET: + 'Optional per-network override for polling interval; shared fallback BLEND_POLL_INTERVAL_MS is documented in .env.example.', // Legacy single-network fallbacks where paired network-specific variables are documented FACILITATOR_FEE_STROOPS: diff --git a/src/__tests__/lendingRateRetention.test.ts b/src/__tests__/lendingRateRetention.test.ts new file mode 100644 index 00000000..7a4541cd --- /dev/null +++ b/src/__tests__/lendingRateRetention.test.ts @@ -0,0 +1,111 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' + +const { mockQuery } = vi.hoisted(() => ({ mockQuery: vi.fn() })) + +vi.mock('../db', () => ({ + prisma: {}, + pgPool: { query: mockQuery }, +})) + +vi.mock('bullmq', () => ({ + Queue: class {}, + Worker: class {}, +})) + +import { + pruneOldRates, + pruneAllRateNetworks, + RATE_PRUNE_BATCH_SIZE, + RATE_RETENTION_DAYS, +} from '../jobs/lendingRateRetention' +import { activeNetwork } from '../config' + +describe('pruneOldRates', () => { + beforeEach(() => { + mockQuery.mockReset() + }) + + it('defaults to the 30-day retention window and bounded batch size', async () => { + mockQuery.mockResolvedValueOnce({ rowCount: 10 }).mockResolvedValueOnce({ rowCount: 0 }) + + const pruned = await pruneOldRates() + + expect(RATE_RETENTION_DAYS).toBe(30) + expect(pruned).toBe(10) + expect(mockQuery.mock.calls[0][1]).toEqual([activeNetwork, 30, RATE_PRUNE_BATCH_SIZE]) + expect(mockQuery.mock.calls[0][0]).toMatch(/DELETE FROM lending_rates AS rates/) + expect(mockQuery.mock.calls[0][0]).toMatch(/LIMIT \$3\s+FOR UPDATE SKIP LOCKED/) + expect(mockQuery.mock.calls[0][0]).toMatch(/timestamp < NOW\(\) - \(\$2 \|\| ' days'\)::interval/) + expect(mockQuery).toHaveBeenCalledTimes(2) + }) + + it('honors a custom retention window', async () => { + mockQuery.mockResolvedValue({ rowCount: 0 }) + + const pruned = await pruneOldRates(7) + + expect(pruned).toBe(0) + expect(mockQuery.mock.calls[0][1]).toEqual([activeNetwork, 7, RATE_PRUNE_BATCH_SIZE]) + }) + + it('binds the requested network instead of the active one', async () => { + mockQuery + .mockResolvedValueOnce({ rowCount: 5 }) + .mockResolvedValueOnce({ rowCount: 0 }) + .mockResolvedValueOnce({ rowCount: 3 }) + .mockResolvedValueOnce({ rowCount: 0 }) + + await pruneOldRates(30, 'mainnet') + await pruneOldRates(30, 'testnet') + + expect(mockQuery.mock.calls[0][1]).toEqual(['mainnet', 30, RATE_PRUNE_BATCH_SIZE]) + expect(mockQuery.mock.calls[2][1]).toEqual(['testnet', 30, RATE_PRUNE_BATCH_SIZE]) + }) + + it('drains multiple bounded batches and reports cumulative progress', async () => { + mockQuery + .mockResolvedValueOnce({ rowCount: 4 }) + .mockResolvedValueOnce({ rowCount: 2 }) + .mockResolvedValueOnce({ rowCount: 0 }) + const log = vi.spyOn(console, 'log').mockImplementation(() => {}) + const onProgress = vi.fn() + + try { + const total = await pruneOldRates(30, 'mainnet', 4, onProgress) + expect(total).toBe(6) + expect(mockQuery.mock.calls.map(call => call[1])).toEqual([ + ['mainnet', 30, 4], + ['mainnet', 30, 4], + ['mainnet', 30, 4], + ]) + expect(onProgress).toHaveBeenLastCalledWith({ + network: 'mainnet', + batch: 2, + batchRows: 2, + totalPruned: 6, + }) + } finally { + log.mockRestore() + } + }) + + it('rejects an invalid batchSize with a RangeError', async () => { + await expect(pruneOldRates(30, 'mainnet', 0)).rejects.toThrow(RangeError) + await expect(pruneOldRates(30, 'mainnet', -5)).rejects.toThrow(RangeError) + await expect(pruneOldRates(30, 'mainnet', NaN)).rejects.toThrow(RangeError) + }) +}) + +describe('pruneAllRateNetworks', () => { + beforeEach(() => { + mockQuery.mockReset() + }) + + it('prunes every enabled network independently', async () => { + mockQuery.mockResolvedValue({ rowCount: 0 }) + + const counts = await pruneAllRateNetworks(30) + expect(typeof counts).toBe('object') + expect(mockQuery).toHaveBeenCalled() + }) +}) diff --git a/src/__tests__/ratesRoute.test.ts b/src/__tests__/ratesRoute.test.ts new file mode 100644 index 00000000..73576140 --- /dev/null +++ b/src/__tests__/ratesRoute.test.ts @@ -0,0 +1,131 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import Fastify from 'fastify' + +const mockRates = [ + { + id: 'rate-1', + network: 'mainnet', + source: 'blend', + pool_id: 'CAJJZSGMMM3PD7N33TAPHGBUGTB43OC73HVIK2L2G6BNGGGYOSSYBXBD', + asset: 'USDC', + asset_address: 'CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75', + supply_apy: 0.052, + borrow_apy: 0.081, + supply_apr: 0.050, + borrow_apr: 0.078, + utilization: 0.65, + total_supply: 1000000, + total_borrowed: 650000, + reserves: 350000, + ledger: 123456, + timestamp: new Date('2026-10-17T10:00:00.000Z'), + }, +] + +const { mockGetLatestLendingRates } = vi.hoisted(() => ({ + mockGetLatestLendingRates: vi.fn(), +})) + +vi.mock('../db', () => ({ + getLatestLendingRates: mockGetLatestLendingRates, +})) + +import { registerRatesRoutes } from '../routes/rates' +import { registerNetworkSelector } from '../middleware/network' + +async function buildTestApp() { + const app = Fastify({ logger: false }) + await app.register(registerNetworkSelector) + await registerRatesRoutes(app) + await app.ready() + return app +} + +describe('Rates API Routes (GET /rates & GET /rates/:asset)', () => { + beforeEach(() => { + mockGetLatestLendingRates.mockReset() + mockGetLatestLendingRates.mockResolvedValue(mockRates) + }) + + it('GET /rates returns lending rates with venue and source as blend', async () => { + const app = await buildTestApp() + const res = await app.inject({ + method: 'GET', + url: '/rates?network=mainnet', + }) + + expect(res.statusCode).toBe(200) + const json = res.json() + expect(json.network).toBe('mainnet') + expect(json.rates).toHaveLength(1) + + const rate = json.rates[0] + expect(rate.venue).toBe('blend') + expect(rate.source).toBe('blend') + expect(rate.asset).toBe('USDC') + expect(rate.poolId).toBe('CAJJZSGMMM3PD7N33TAPHGBUGTB43OC73HVIK2L2G6BNGGGYOSSYBXBD') + expect(rate.supplyApy).toBe(0.052) + expect(rate.borrowApy).toBe(0.081) + expect(rate.utilization).toBe(0.65) + expect(rate.reserves).toBe(350000) + + expect(mockGetLatestLendingRates).toHaveBeenCalledWith('mainnet', { + asset: undefined, + poolId: undefined, + limit: 100, + }) + + await app.close() + }) + + it('GET /rates/:asset filters for a specific asset', async () => { + const app = await buildTestApp() + const res = await app.inject({ + method: 'GET', + url: '/rates/USDC?network=mainnet', + }) + + expect(res.statusCode).toBe(200) + const json = res.json() + expect(json.asset).toBe('USDC') + expect(json.network).toBe('mainnet') + expect(json.rates[0].asset).toBe('USDC') + + expect(mockGetLatestLendingRates).toHaveBeenCalledWith('mainnet', { + asset: 'USDC', + poolId: undefined, + limit: 100, + }) + + await app.close() + }) + + it('GET /rates passes poolId and limit filters', async () => { + const app = await buildTestApp() + const poolId = 'CAJJZSGMMM3PD7N33TAPHGBUGTB43OC73HVIK2L2G6BNGGGYOSSYBXBD' + const res = await app.inject({ + method: 'GET', + url: `/rates?poolId=${poolId}&limit=50&network=mainnet`, + }) + + expect(res.statusCode).toBe(200) + expect(mockGetLatestLendingRates).toHaveBeenCalledWith('mainnet', { + asset: undefined, + poolId, + limit: 50, + }) + + await app.close() + }) + + it('rejects an invalid network parameter with 400', async () => { + const app = await buildTestApp() + const res = await app.inject({ + method: 'GET', + url: '/rates?network=invalid_net', + }) + + expect(res.statusCode).toBe(400) + await app.close() + }) +}) diff --git a/src/app.ts b/src/app.ts index c0d2a117..2994fe8d 100644 --- a/src/app.ts +++ b/src/app.ts @@ -26,6 +26,7 @@ import { registerOracleRoutes } from './routes/oracle' import { registerBasketRoutes } from './routes/basket' import { registerDiscoveryRoutes } from './routes/discovery' import { registerSettleRoute } from './routes/facilitator' +import { registerRatesRoutes } from './routes/rates' import { getMetrics } from './metrics' export interface BuildAppOptions { @@ -139,6 +140,7 @@ export async function buildApp(options: BuildAppOptions = {}): Promise p.trim()) + .filter(p => isUsableContractId(p, `BLEND_POOLS_${suffix}`)) + : (network === 'mainnet' ? MAINNET_BLEND_POOLS : TESTNET_BLEND_POOLS) + + const blendEnabled = + (process.env[`BLEND_ENABLED_${suffix}`] ?? (network === 'testnet' ? (process.env.BLEND_ENABLED ?? 'true') : 'true')).toLowerCase() !== 'false' && + parsedBlendPools.length > 0 && + isUsableContractId(blendBackstop, `BLEND_BACKSTOP_${suffix}`) + + const blendPollMs = parseInt( + process.env[`BLEND_POLL_INTERVAL_MS_${suffix}`] || + process.env.BLEND_POLL_INTERVAL_MS || + '60000', + 10 + ) + // ── Watched pairs ───────────────────────────────────────────────────────── // Mainnet gets a default for the same reason every field above does: turning // it on should not require knowing an issuer address by heart. USDC/XLM is @@ -225,6 +277,12 @@ function buildNetworkConfig(network: NetworkName): NetworkConfig { apiUrl: aquariusApiUrl, }, oracle: { enabled: oracleEnabled, reflectorContractId }, + blend: { + enabled: blendEnabled, + pools: parsedBlendPools, + backstopContractId: blendBackstop, + pollIntervalMs: blendPollMs, + }, pairs: parseWatchedPairs(rawPairs), facilitator: { secretKey: facilitatorSecretKey, feeStroops: facilitatorFeeStroops }, } @@ -324,6 +382,7 @@ export const config = { get soroswap() { return resolveNetwork(activeNetwork).soroswap }, get aquarius() { return resolveNetwork(activeNetwork).aquarius }, get oracle() { return resolveNetwork(activeNetwork).oracle }, + get blend() { return resolveNetwork(activeNetwork).blend }, get pairs() { return resolveNetwork(activeNetwork).pairs }, get facilitator() { return resolveNetwork(activeNetwork).facilitator }, diff --git a/src/db.ts b/src/db.ts index a98b66cc..1e7a9b1a 100644 --- a/src/db.ts +++ b/src/db.ts @@ -85,3 +85,77 @@ export async function setIndexerCursor(id: string, cursor: string, network: Netw update: { lastCursor: cursor, lastLedger: ledger, lastProcessedAt: new Date() }, }) } + +export interface LendingRateInput { + poolId: string + asset: string + assetAddress?: string + source?: string + supplyApy: number + borrowApy: number + supplyApr?: number + borrowApr?: number + utilization: number + totalSupply?: number + totalBorrowed?: number + reserves?: number + ledger?: number + timestamp?: Date +} + +export async function insertLendingRates( + rates: LendingRateInput[], + network: NetworkName +): Promise { + if (rates.length === 0) return 0 + const result = await prisma.lendingRate.createMany({ + data: rates.map(r => ({ + network, + source: r.source ?? 'blend', + poolId: r.poolId, + asset: r.asset, + assetAddress: r.assetAddress ?? null, + supplyApy: r.supplyApy, + borrowApy: r.borrowApy, + supplyApr: r.supplyApr ?? null, + borrowApr: r.borrowApr ?? null, + utilization: r.utilization, + totalSupply: r.totalSupply ?? null, + totalBorrowed: r.totalBorrowed ?? null, + reserves: r.reserves ?? null, + ledger: r.ledger ?? 0, + timestamp: r.timestamp ?? new Date(), + })), + }) + return result.count +} + +export async function getLatestLendingRates( + network: NetworkName, + options?: { asset?: string; poolId?: string; limit?: number } +) { + const limit = options?.limit ?? 100 + let query = ` + SELECT DISTINCT ON (pool_id, asset) + id, network, source, pool_id, asset, asset_address, + supply_apy::float, borrow_apy::float, supply_apr::float, borrow_apr::float, + utilization::float, total_supply::float, total_borrowed::float, reserves::float, + ledger, timestamp + FROM lending_rates + WHERE network = $1 + ` + const params: any[] = [network] + if (options?.asset) { + params.push(options.asset.toUpperCase()) + query += ` AND UPPER(asset) = $${params.length}` + } + if (options?.poolId) { + params.push(options.poolId) + query += ` AND pool_id = $${params.length}` + } + query += ` ORDER BY pool_id, asset, timestamp DESC LIMIT $${params.length + 1}` + params.push(limit) + + const result = await pgPool.query(query, params) + return result.rows +} diff --git a/src/index.ts b/src/index.ts index 18f372c4..1100389e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -11,10 +11,11 @@ import { pgPool } from './db' import { buildApp } from './app' import { fanOutManager } from './ws/fanout' -import { venueAdapters, startVenueAdapter } from './ingest/venues' +import { venueAdapters, startVenueAdapter, startBlendIngester } from './ingest/venues' import { startSnapshotIngester } from './ingesters/snapshot' import { createAggregateQueue, startAggregateWorker, scheduleAggregateRefresh } from './jobs/aggregateRefresh' import { createSnapshotRetentionQueue, startSnapshotRetentionWorker, scheduleSnapshotRetention, pruneAllNetworks, SNAPSHOT_RETENTION_DAYS } from './jobs/snapshotRetention' +import { createRateRetentionQueue, startRateRetentionWorker, scheduleRateRetention, pruneAllRateNetworks, RATE_RETENTION_DAYS } from './jobs/lendingRateRetention' import { loadPersistedPairs, getActivePairs } from './pairsRegistry' async function main() { @@ -135,6 +136,38 @@ async function main() { } } + // ── Lending rate retention ──────────────────────────────────────────────── + const safeRatePrune = async () => { + try { + const counts = await pruneAllRateNetworks() + for (const [network, pruned] of Object.entries(counts)) { + if (pruned > 0) { + console.log(`[lens] Pruned ${pruned} ${network} lending rate(s) older than ${RATE_RETENTION_DAYS}d`) + } + } + } catch (err) { + console.error('[lens] Lending rate prune failed:', (err as Error).message) + } + } + + await safeRatePrune() + + { + let rateRetentionQueue: ReturnType | undefined + let rateRetentionWorker: ReturnType | undefined + try { + rateRetentionQueue = createRateRetentionQueue() + rateRetentionWorker = startRateRetentionWorker() + await scheduleRateRetention(rateRetentionQueue) + console.log('[lens] Lending rate retention worker started') + } catch (err) { + console.warn('[lens] Lending rate retention worker skipped (Redis unavailable):', (err as Error).message) + console.warn('[lens] Falling back to an in-process hourly rate prune') + await Promise.allSettled([rateRetentionQueue?.close(), rateRetentionWorker?.close()]) + setInterval(() => { void safeRatePrune() }, 60 * 60 * 1000).unref() + } + } + // ── Ingesters (run in background — infinite loops) ──────────────────────── // Each ingester is independently fault-isolated via restartIngester, keyed by // the (venue, network) pair: a crash in, say, the Soroswap ingester on @@ -154,6 +187,7 @@ async function main() { restartIngester(adapter.name, network, () => startVenueAdapter(adapter, network)) } restartIngester('Snapshot', network, () => startSnapshotIngester(network)) + restartIngester('Blend', network, () => startBlendIngester(network)) } console.log(`[lens] Watching ${getActivePairs().length} pairs: ${getActivePairs().map(p => p.pairKey).join(', ')}`) diff --git a/src/ingest/venues/blend.ts b/src/ingest/venues/blend.ts new file mode 100644 index 00000000..098e9826 --- /dev/null +++ b/src/ingest/venues/blend.ts @@ -0,0 +1,632 @@ +/** + * Blend Lending Protocol Venue Adapter & Rate Ingester + * + * Reads Blend lending pool reserves, calculates derived utilisation, + * supply APY, and borrow APY, and stores them in the lending_rates table. + * + * Unlike DEX venues, lending rates are not spot prices and are stored + * in their own table (lending_rates) rather than price_points. + */ + +import { + Address, + xdr, + scValToNative, + Contract, + TransactionBuilder, + BASE_FEE, + Networks, + Account, + Keypair, + rpc as SorobanRpc, +} from '@stellar/stellar-sdk' +import { getNetworkConfig, activeNetwork, type NetworkName } from '../../config' +import { getRpcServer } from '../../network/clients' +import { insertLendingRates, type LendingRateInput } from '../../db' +import { blend_pool_errors_total, blend_rates_ingested_total } from '../../metrics' +import type { WatchedPair } from '../../types' +import type { VenueAdapter, VenuePricePoint } from './types' +import { getLastPrice, resetLastPrice, sleep } from './runner' + +// ── Ephemeral Fee Payer for Soroban Simulations ─────────────────────────────── +const FEE_PAYER_KEYPAIR = Keypair.random() + +// ── In-memory caches ───────────────────────────────────────────────────────── +const assetSymbolCache = new Map() + +// Common known Stellar tokens on Mainnet and Testnet +const KNOWN_ASSET_SYMBOLS: Record = { + // Mainnet + CAS3J7GYLGXMF6TDJBBYYSE3HQ6BBSMLNUQ34T6TZMYMW2EZH3MS2UY7: 'XLM', + CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC: 'XLM', + CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75: 'USDC', + CD252NPIMBFLRLNYWVDZEPISECT7NWGCVME5BC2Z5AOZ2MQ27AC7T5NX: 'BLND', + CA4NZURNP4MWX5CW74WPUEX44WJ4PB7BR4A362XW3J24TF3J77Y7H6K7: 'wETH', + // Testnet + CBIELTK6YBZJU5UP2WWQEUCYKLPU6AUNZ2BQ4WWUIE3USOPWWYPYYOB7: 'USDC', +} + +// ── Test helper hooks matching reference adapter structure ─────────────────── +export function _resetLastPrice(): void { + resetLastPrice('Blend') +} + +export function _getLastPrice(network: NetworkName, pairKey: string): number | undefined { + return getLastPrice('Blend', network, pairKey) +} + +// ── Types ───────────────────────────────────────────────────────────────────── + +export interface BlendReserveConfig { + index: number + decimals: number + c_factor?: number | bigint + l_factor?: number | bigint + util: number | bigint + max_util?: number | bigint + r_base: number | bigint + r_one: number | bigint + r_two: number | bigint + r_three: number | bigint + reactivity?: number | bigint + supply_cap?: bigint + enabled?: boolean +} + +export interface BlendReserveData { + b_rate?: bigint + bRate?: bigint + b_supply?: bigint + bSupply?: bigint + d_rate?: bigint + dRate?: bigint + d_supply?: bigint + dSupply?: bigint + ir_mod?: bigint + interestRateModifier?: bigint + backstop_credit?: bigint + backstopCredit?: bigint + last_time?: number + lastTime?: number +} + +export interface BlendReserveEntry { + assetId: string + assetSymbol: string + config: BlendReserveConfig + data: BlendReserveData + rateDecimals: number +} + +export interface BlendPoolData { + poolId: string + network: NetworkName + backstopRate: bigint + reserves: BlendReserveEntry[] +} + +export interface DerivedRates { + utilization: number + borrowApr: number + supplyApr: number + borrowApy: number + supplyApy: number + totalSupply: number + totalBorrowed: number + reserves: number +} + +// ── Mathematical Calculations ──────────────────────────────────────────────── + +/** + * Derives utilisation as a fraction in [0, 1]. + * Formula: totalLiabilities / totalSupply. + * Pins arithmetic: + * - If liabilities <= 0 or supply <= 0 => 0 + * - If liabilities >= supply => 1.0 + * - Otherwise => Number(liabilities) / Number(supply) + */ +export function calcUtilization( + liabilities: bigint | number, + supply: bigint | number +): number { + const liabNum = typeof liabilities === 'bigint' ? Number(liabilities) : liabilities + const suppNum = typeof supply === 'bigint' ? Number(supply) : supply + + if (liabNum <= 0 || suppNum <= 0 || !Number.isFinite(liabNum) || !Number.isFinite(suppNum)) { + return 0 + } + if (liabNum >= suppNum) { + return 1.0 + } + return liabNum / suppNum +} + +/** + * Calculates liabilities and supply from raw b/d tokens and derives utilisation. + * In Blend: + * totalLiabilities = ceil(dSupply * dRate / 10^rateDecimals) + * totalSupply = floor(bSupply * bRate / 10^rateDecimals) + */ +export function deriveUtilization( + dSupply: bigint, + dRate: bigint, + bSupply: bigint, + bRate: bigint, + rateDecimals: number = 12 +): number { + const scalar = 10n ** BigInt(rateDecimals) + // mulDivCeil(dSupply, dRate, scalar) + const dProd = dSupply * dRate + const liabilities = dProd / scalar + (dProd % scalar > 0n ? 1n : 0n) + + // mulDivFloor(bSupply, bRate, scalar) + const supply = (bSupply * bRate) / scalar + + return calcUtilization(liabilities, supply) +} + +/** + * Calculates APR and APY for borrowers and suppliers from Blend's 3-tier interest rate curve. + * + * Curve specs (scaled by 1e7): + * - Target utilisation: config.util (usually ~75%) + * - 95% kink: fixed 95% + * - Rates: r_base + piecewise slopes (r_one, r_two, r_three) modulated by interestRateModifier + * - APY compounded: Borrowing daily (365x), Supplying weekly (52x) + */ +export function calcInterestRates( + config: BlendReserveConfig, + data: BlendReserveData, + utilization: number, + backstopRate: bigint | number = 1_000_000n // default 10% (1e6 / 1e7) +): { borrowApr: number; supplyApr: number; borrowApy: number; supplyApy: number } { + const SCALAR_7 = 10_000_000 + + const targetUtil = Number(config.util) / SCALAR_7 + const rBase = Number(config.r_base) / SCALAR_7 + const rOne = Number(config.r_one) / SCALAR_7 + const rTwo = Number(config.r_two) / SCALAR_7 + const rThree = Number(config.r_three) / SCALAR_7 + + const rawIrMod = data.ir_mod ?? data.interestRateModifier ?? 10_000_000n + const irMod = Number(rawIrMod) / SCALAR_7 + + const bstopTake = (typeof backstopRate === 'bigint' ? Number(backstopRate) : backstopRate) / SCALAR_7 + + let borrowApr: number + if (utilization <= targetUtil) { + const utilScalar = targetUtil === 0 ? 0 : utilization / targetUtil + const baseRate = utilScalar * rOne + rBase + borrowApr = baseRate * irMod + } else if (utilization <= 0.95) { + const denom = 0.95 - targetUtil + const utilScalar = denom === 0 ? 0 : (utilization - targetUtil) / denom + const baseRate = utilScalar * rTwo + rOne + rBase + borrowApr = baseRate * irMod + } else { + const utilScalar = (utilization - 0.95) / 0.05 + const extraRate = utilScalar * rThree + const intersection = irMod * (rTwo + rOne + rBase) + borrowApr = extraRate + intersection + } + + // Supply capture: (1 - backstopTakeRate) * utilization + const supplyCapture = Math.max(0, 1 - bstopTake) * utilization + const supplyApr = borrowApr * supplyCapture + + // Daily compounding for borrow APY: (1 + borrowApr / 365) ** 365 - 1 + const borrowApy = Math.pow(1 + borrowApr / 365, 365) - 1 + // Weekly compounding for supply APY: (1 + supplyApr / 52) ** 52 - 1 + const supplyApy = Math.pow(1 + supplyApr / 52, 52) - 1 + + return { + borrowApr: Number.isFinite(borrowApr) ? borrowApr : 0, + supplyApr: Number.isFinite(supplyApr) ? supplyApr : 0, + borrowApy: Number.isFinite(borrowApy) ? borrowApy : 0, + supplyApy: Number.isFinite(supplyApy) ? supplyApy : 0, + } +} + +/** + * Derives full rate metrics for a reserve entry. + */ +export function deriveReserveRates( + reserve: BlendReserveEntry, + backstopRate: bigint +): DerivedRates { + const dSupply = reserve.data.d_supply ?? reserve.data.dSupply ?? 0n + const dRate = reserve.data.d_rate ?? reserve.data.dRate ?? 10n ** BigInt(reserve.rateDecimals) + const bSupply = reserve.data.b_supply ?? reserve.data.bSupply ?? 0n + const bRate = reserve.data.b_rate ?? reserve.data.bRate ?? 10n ** BigInt(reserve.rateDecimals) + + const scalar = 10n ** BigInt(reserve.rateDecimals) + const dProd = dSupply * dRate + const totalLiabilities = dProd / scalar + (dProd % scalar > 0n ? 1n : 0n) + const totalSupply = (bSupply * bRate) / scalar + + const utilization = calcUtilization(totalLiabilities, totalSupply) + const rates = calcInterestRates(reserve.config, reserve.data, utilization, backstopRate) + + const assetDecimals = reserve.config.decimals ?? 7 + const assetDivisor = 10 ** assetDecimals + + const totalSupplyFloat = Number(totalSupply) / assetDivisor + const totalBorrowedFloat = Number(totalLiabilities) / assetDivisor + const reservesFloat = Math.max(0, totalSupplyFloat - totalBorrowedFloat) + + return { + utilization, + borrowApr: rates.borrowApr, + supplyApr: rates.supplyApr, + borrowApy: rates.borrowApy, + supplyApy: rates.supplyApy, + totalSupply: totalSupplyFloat, + totalBorrowed: totalBorrowedFloat, + reserves: reservesFloat, + } +} + +// ── Asset Symbol Resolution ────────────────────────────────────────────────── + +/** + * Resolves token contract address to human-readable symbol (e.g. 'USDC', 'XLM'). + */ +export async function resolveAssetSymbol( + assetId: string, + network: NetworkName +): Promise { + if (KNOWN_ASSET_SYMBOLS[assetId]) { + return KNOWN_ASSET_SYMBOLS[assetId] + } + + const cached = assetSymbolCache.get(assetId) + if (cached) return cached + + try { + const rpc = getRpcServer(network) + const contract = new Contract(assetId) + const account = new Account(FEE_PAYER_KEYPAIR.publicKey(), '0') + const networkPassphrase = + getNetworkConfig(network).network.passphrase ?? Networks.PUBLIC + + const tx = new TransactionBuilder(account, { + fee: BASE_FEE, + networkPassphrase, + }) + .addOperation(contract.call('symbol')) + .setTimeout(30) + .build() + + const sim = await rpc.simulateTransaction(tx) + if (SorobanRpc.Api.isSimulationSuccess(sim) && sim.result?.retval) { + const sym = scValToNative(sim.result.retval) + if (typeof sym === 'string' && sym.length > 0) { + assetSymbolCache.set(assetId, sym) + return sym + } + } + } catch { + // Contract simulation failed — fallback below + } + + // Fallback to shortened contract address + const fallback = `${assetId.slice(0, 4)}...${assetId.slice(-4)}` + assetSymbolCache.set(assetId, fallback) + return fallback +} + +// ── Pool State Ingestion ───────────────────────────────────────────────────── + +/** + * Fetches raw Blend pool data from Soroban RPC. + * Returns null if the pool cannot be loaded. + */ +export async function fetchBlendPool( + poolId: string, + network: NetworkName +): Promise { + const rpc = getRpcServer(network) + + const contractInstanceKey = xdr.LedgerKey.contractData( + new xdr.LedgerKeyContractData({ + contract: Address.fromString(poolId).toScAddress(), + key: xdr.ScVal.scvLedgerKeyContractInstance(), + durability: xdr.ContractDataDurability.persistent(), + }) + ) + + const resListKey = xdr.LedgerKey.contractData( + new xdr.LedgerKeyContractData({ + contract: Address.fromString(poolId).toScAddress(), + key: xdr.ScVal.scvSymbol('ResList'), + durability: xdr.ContractDataDurability.persistent(), + }) + ) + + const initialEntries = await rpc.getLedgerEntries(contractInstanceKey, resListKey) + if (!initialEntries.entries || initialEntries.entries.length === 0) { + return null + } + + let reserveList: string[] = [] + let backstopRate = 1_000_000n // default 10% + + for (const entry of initialEntries.entries) { + const cd = entry.val.contractData() + const keyScVal = cd.key() + + if (keyScVal.switch() === xdr.ScValType.scvSymbol() && keyScVal.sym().toString() === 'ResList') { + const nativeList = scValToNative(cd.val()) + if (Array.isArray(nativeList)) { + reserveList = nativeList.map((addr: any) => + typeof addr === 'string' ? addr : addr.toString() + ) + } + } else if (keyScVal.switch() === xdr.ScValType.scvLedgerKeyContractInstance()) { + const instance = cd.val().instance() + const storage = instance.storage() + if (storage) { + for (const storageEntry of storage) { + const k = storageEntry.key() + if (k.switch() === xdr.ScValType.scvSymbol() && k.sym().toString() === 'Config') { + const configObj = scValToNative(storageEntry.val()) + if (configObj && (configObj.bstop_rate !== undefined || configObj.backstopRate !== undefined)) { + const rateVal = configObj.bstop_rate ?? configObj.backstopRate + backstopRate = BigInt(rateVal) + } + } + } + } + } + } + + if (reserveList.length === 0) { + return null + } + + // Fetch reserve configs and reserve data for all assets in the pool + const ledgerKeys: xdr.LedgerKey[] = [] + for (const assetId of reserveList) { + ledgerKeys.push( + xdr.LedgerKey.contractData( + new xdr.LedgerKeyContractData({ + contract: Address.fromString(poolId).toScAddress(), + key: xdr.ScVal.scvVec([ + xdr.ScVal.scvSymbol('ResConfig'), + Address.fromString(assetId).toScVal(), + ]), + durability: xdr.ContractDataDurability.persistent(), + }) + ) + ) + ledgerKeys.push( + xdr.LedgerKey.contractData( + new xdr.LedgerKeyContractData({ + contract: Address.fromString(poolId).toScAddress(), + key: xdr.ScVal.scvVec([ + xdr.ScVal.scvSymbol('ResData'), + Address.fromString(assetId).toScVal(), + ]), + durability: xdr.ContractDataDurability.persistent(), + }) + ) + ) + } + + const reserveEntries = await rpc.getLedgerEntries(...ledgerKeys) + const configsMap = new Map() + const dataMap = new Map() + + for (const entry of reserveEntries.entries ?? []) { + const cd = entry.val.contractData() + const k = cd.key() + if (k.switch() === xdr.ScValType.scvVec()) { + const vec = k.vec() + if (!vec || vec.length < 2) continue + const tag = vec[0].sym().toString() + const assetAddr = Address.fromScVal(vec[1]).toString() + const val = scValToNative(cd.val()) + + if (tag === 'ResConfig') { + configsMap.set(assetAddr, val as BlendReserveConfig) + } else if (tag === 'ResData') { + dataMap.set(assetAddr, val as BlendReserveData) + } + } + } + + const reserves: BlendReserveEntry[] = [] + for (const assetId of reserveList) { + const config = configsMap.get(assetId) + const data = dataMap.get(assetId) + if (!config || !data) continue + + const assetSymbol = await resolveAssetSymbol(assetId, network) + reserves.push({ + assetId, + assetSymbol, + config, + data, + rateDecimals: 12, // Blend v2 uses 12 decimals + }) + } + + return { + poolId, + network, + backstopRate, + reserves, + } +} + +/** + * Ingests a single Blend pool. + * If the pool cannot be read, it is skipped with a log and a metric increment, NOT a crash. + */ +export async function ingestBlendPool( + poolId: string, + network: NetworkName, + fetchPool = fetchBlendPool +): Promise { + try { + const pool = await fetchPool(poolId, network) + if (!pool || pool.reserves.length === 0) { + console.warn(`[blend] Skipping pool ${poolId} on ${network}: no reserves found or pool could not be read`) + blend_pool_errors_total.inc({ pool: poolId, network }) + return [] + } + + const rates: LendingRateInput[] = [] + const now = new Date() + + for (const reserve of pool.reserves) { + const derived = deriveReserveRates(reserve, pool.backstopRate) + + rates.push({ + poolId, + asset: reserve.assetSymbol, + assetAddress: reserve.assetId, + source: 'blend', + supplyApy: derived.supplyApy, + borrowApy: derived.borrowApy, + supplyApr: derived.supplyApr, + borrowApr: derived.borrowApr, + utilization: derived.utilization, + totalSupply: derived.totalSupply, + totalBorrowed: derived.totalBorrowed, + reserves: derived.reserves, + timestamp: now, + }) + } + + if (rates.length > 0) { + await insertLendingRates(rates, network) + blend_rates_ingested_total.inc({ pool: poolId, network }, rates.length) + } + + return rates + } catch (err) { + console.error(`[blend] Failed to ingest pool ${poolId} on ${network}:`, (err as Error).message) + blend_pool_errors_total.inc({ pool: poolId, network }) + return [] + } +} + +/** + * Ingests all configured Blend pools for a network. + */ +export async function ingestBlendPools( + network: NetworkName, + ingestPool = ingestBlendPool +): Promise { + const netConfig = getNetworkConfig(network) + if (!netConfig.blend.enabled) { + return 0 + } + + let totalIngested = 0 + for (const poolId of netConfig.blend.pools) { + try { + const rates = await ingestPool(poolId, network) + totalIngested += rates.length + } catch (err) { + console.error(`[blend] Failed to ingest pool ${poolId} on ${network}:`, (err as Error).message) + blend_pool_errors_total.inc({ pool: poolId, network }) + } + } + return totalIngested +} + +/** + * Ingests blend pools matching a pair if applicable. + */ +export async function ingestBlendPair( + pair: WatchedPair, + network: NetworkName = activeNetwork +): Promise { + await ingestBlendPools(network) +} + +// ── VenueAdapter Contract Implementation ────────────────────────────────────── + +export const blendAdapter: VenueAdapter = { + name: 'Blend', + source: 'blend', + + isEnabled(network: NetworkName): boolean { + return getNetworkConfig(network).blend.enabled + }, + + getPollIntervalMs(network: NetworkName): number { + return getNetworkConfig(network).blend.pollIntervalMs + }, + + async discover(_pair: WatchedPair, network: NetworkName): Promise { + const netConfig = getNetworkConfig(network) + if (!netConfig.blend.enabled) return [] + const pools: BlendPoolData[] = [] + for (const poolId of netConfig.blend.pools) { + const data = await fetchBlendPool(poolId, network) + if (data) pools.push(data) + } + return pools.length > 0 ? [pools] : [] + }, + + async read(target: BlendPoolData[], _pair: WatchedPair, _network: NetworkName): Promise { + return target.flatMap((p) => p.reserves) + }, + + derive(_raw: BlendReserveEntry[], _target: BlendPoolData[], _pair: WatchedPair, _network: NetworkName): VenuePricePoint[] { + // Blend rates are stored in lending_rates, not price_points. + return [] + }, + + async write(_points: VenuePricePoint[], target: BlendPoolData[], _pair: WatchedPair, network: NetworkName): Promise { + for (const pool of target) { + await ingestBlendPool(pool.poolId, network) + } + }, + + async ingestPair(pair: WatchedPair, network: NetworkName): Promise { + await ingestBlendPair(pair, network) + }, + + _resetLastPrice, + _getLastPrice, +} + +/** + * Starts the Blend ingester polling loop for a network. + */ +export async function startBlendIngester( + network: NetworkName = activeNetwork +): Promise { + const netConfig = getNetworkConfig(network) + if (!netConfig.blend.enabled) { + console.log(`[blend] Blend is disabled on ${network} — ingester not started`) + return + } + + console.log(`[blend] Starting Blend lending rate ingester on ${network} (pools: ${netConfig.blend.pools.length})`) + + let consecutiveErrors = 0 + while (true) { + try { + await ingestBlendPools(network) + consecutiveErrors = 0 + } catch (err) { + consecutiveErrors++ + const backoffMs = Math.min(10_000 * Math.pow(2, consecutiveErrors - 1), 60_000) + console.error( + `[blend] Ingestion cycle failed on ${network} (attempt ${consecutiveErrors}), retrying in ${backoffMs}ms:`, + (err as Error).message + ) + await sleep(backoffMs) + continue + } + + const pollInterval = netConfig.blend.pollIntervalMs + await sleep(pollInterval) + } +} diff --git a/src/ingest/venues/index.ts b/src/ingest/venues/index.ts index 0ea72a28..db5e5c64 100644 --- a/src/ingest/venues/index.ts +++ b/src/ingest/venues/index.ts @@ -10,6 +10,7 @@ export { sdexAdapter, startSDEXIngester } from './sdex' export { ammAdapter, startAMMIngester } from './amm' export { soroswapAdapter, startSoroswapIngester } from './soroswap' export { aquariusAdapter, startAquariusIngester } from './aquarius' +export { blendAdapter, startBlendIngester } from './blend' /** * Registry of all available venue adapters. diff --git a/src/jobs/lendingRateRetention.ts b/src/jobs/lendingRateRetention.ts new file mode 100644 index 00000000..8d214b3c --- /dev/null +++ b/src/jobs/lendingRateRetention.ts @@ -0,0 +1,131 @@ +import { Queue, Worker } from 'bullmq' +import { pgPool } from '../db' +import { activeNetwork, type NetworkName } from '../config' +import { getEnabledNetworks } from '../network/enabledNetworks' + +const QUEUE_NAME = `${activeNetwork}:rate-retention` + +/** Lending rates older than this many days are pruned by the retention job. */ +export const RATE_RETENTION_DAYS = 30 +/** Maximum rows removed by one database statement during a retention run. */ +export const RATE_PRUNE_BATCH_SIZE = Number(process.env.RATE_PRUNE_BATCH_SIZE ?? 1_000) + +export interface RatePruneProgress { + network: NetworkName + batch: number + batchRows: number + totalPruned: number +} + +function redisConnection() { + const url = process.env.REDIS_URL + if (url) return { url } + return { host: 'localhost', port: 6379 } +} + +export function createRateRetentionQueue() { + return new Queue(QUEUE_NAME, { connection: redisConnection() }) +} + +/** + * Deletes one network's old lending_rates rows in bounded batches. Locked + * rows are skipped so concurrent retention workers can make progress without + * contending for the same batch. Returns the total number of rows pruned. + */ +export async function pruneOldRates( + retentionDays: number = RATE_RETENTION_DAYS, + network: NetworkName = activeNetwork, + batchSize: number = RATE_PRUNE_BATCH_SIZE, + onProgress?: (progress: RatePruneProgress) => void +): Promise { + if (!Number.isSafeInteger(batchSize) || batchSize <= 0) { + throw new RangeError('Rate prune batch size must be a positive integer') + } + + let totalPruned = 0 + let batchNumber = 0 + + while (true) { + const result = await pgPool.query( + `WITH batch AS ( + SELECT ctid + FROM lending_rates + WHERE network = $1 AND timestamp < NOW() - ($2 || ' days')::interval + ORDER BY timestamp + LIMIT $3 + FOR UPDATE SKIP LOCKED + ) + DELETE FROM lending_rates AS rates + USING batch + WHERE rates.ctid = batch.ctid`, + [network, retentionDays, batchSize] + ) + const pruned = result.rowCount ?? 0 + if (pruned === 0) break + + totalPruned += pruned + batchNumber += 1 + console.log(`[rate-retention] ${network}: pruned batch ${batchNumber} (${pruned} rows; ${totalPruned} total)`) + onProgress?.({ network, batch: batchNumber, batchRows: pruned, totalPruned }) + + // Give other jobs and requests a chance to run between database batches. + await new Promise(resolve => setImmediate(resolve)) + } + + return totalPruned +} + +/** + * Prunes every network in `getEnabledNetworks()` and returns the pruned row + * count per network. + */ +export async function pruneAllRateNetworks( + retentionDays: number = RATE_RETENTION_DAYS +): Promise>> { + const counts: Partial> = {} + const errors: Error[] = [] + for (const network of getEnabledNetworks()) { + try { + counts[network] = await pruneOldRates(retentionDays, network) + } catch (err) { + errors.push(new Error(`${network}: ${(err as Error).message}`)) + } + } + if (errors.length > 0) { + throw new AggregateError(errors, errors.map(e => e.message).join('; ')) + } + return counts +} + +export function startRateRetentionWorker() { + const worker = new Worker( + QUEUE_NAME, + async () => { + try { + const counts = await pruneAllRateNetworks() + for (const [network, pruned] of Object.entries(counts)) { + if (pruned > 0) console.log(`[rate-retention] pruned ${pruned} ${network} rate(s) older than ${RATE_RETENTION_DAYS}d`) + } + } catch (err) { + console.error('[rate-retention] prune failed:', (err as Error).message) + } + }, + { connection: redisConnection(), concurrency: 1 } + ) + + worker.on('failed', (_job, err) => { + console.error('[rate-retention] Job failed:', err.message) + }) + + return worker +} + +/** Schedules the rate retention prune to run hourly (and once on startup). */ +export async function scheduleRateRetention(queue: Queue) { + await queue.add( + 'prune', + {}, + { repeat: { every: 60 * 60 * 1000 }, jobId: 'rate-retention:prune' } + ) + await queue.add('prune', {}) +} diff --git a/src/metrics.ts b/src/metrics.ts index 39a3a505..a0ead797 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -40,6 +40,20 @@ export const price_snapshots_total = new Counter({ registers: [register] }) +export const blend_pool_errors_total = new Counter({ + name: 'blend_pool_errors_total', + help: 'Total number of Blend pool read failures', + labelNames: ['pool', 'network'], + registers: [register], +}) + +export const blend_rates_ingested_total = new Counter({ + name: 'blend_rates_ingested_total', + help: 'Total number of Blend pool rate records ingested', + labelNames: ['pool', 'network'], + registers: [register], +}) + export const price_requests_total = new Counter({ name: 'price_requests_total', help: 'Total number of price API requests served', diff --git a/src/routes/rates.ts b/src/routes/rates.ts new file mode 100644 index 00000000..17549bfa --- /dev/null +++ b/src/routes/rates.ts @@ -0,0 +1,108 @@ +import type { FastifyInstance } from 'fastify' +import { z } from 'zod' +import { getLatestLendingRates } from '../db' +import { activeNetwork, type NetworkName } from '../config' +import '../middleware/network' + +const RatesQuerySchema = z.object({ + asset: z.string().optional(), + poolId: z.string().optional(), + network: z.enum(['testnet', 'mainnet']).optional(), + limit: z.coerce.number().int().min(1).max(1000).default(100), +}) + +/** + * Register Blend lending rate querying endpoints. + * + * GET /rates - Query latest lending rates across pools + * GET /rates/:asset - Query latest lending rates for a specific asset (e.g. USDC, XLM) + */ +export async function registerRatesRoutes(app: FastifyInstance) { + app.get<{ + Querystring: { + asset?: string + poolId?: string + network?: string + limit?: string | number + } + }>('/rates', async (req, reply) => { + const parsed = RatesQuerySchema.safeParse(req.query) + if (!parsed.success) { + return reply.status(400).send({ error: parsed.error.issues[0].message || 'Invalid parameters' }) + } + const network: NetworkName = req.network ?? parsed.data.network ?? activeNetwork + const rows = await getLatestLendingRates(network, { + asset: parsed.data.asset, + poolId: parsed.data.poolId, + limit: parsed.data.limit, + }) + + return { + network, + rates: rows.map(r => ({ + id: r.id, + venue: r.source ?? 'blend', + source: r.source ?? 'blend', + network: r.network, + poolId: r.pool_id, + asset: r.asset, + assetAddress: r.asset_address, + supplyApy: Number(r.supply_apy), + borrowApy: Number(r.borrow_apy), + supplyApr: r.supply_apr !== null ? Number(r.supply_apr) : null, + borrowApr: r.borrow_apr !== null ? Number(r.borrow_apr) : null, + utilization: Number(r.utilization), + totalSupply: r.total_supply !== null ? Number(r.total_supply) : null, + totalBorrowed: r.total_borrowed !== null ? Number(r.total_borrowed) : null, + reserves: r.reserves !== null ? Number(r.reserves) : null, + ledger: r.ledger, + timestamp: r.timestamp instanceof Date ? r.timestamp.toISOString() : r.timestamp, + })), + } + }) + + app.get<{ + Params: { asset: string } + Querystring: { + poolId?: string + network?: string + limit?: string | number + } + }>('/rates/:asset', async (req, reply) => { + const { asset } = req.params + const parsed = RatesQuerySchema.safeParse(req.query) + if (!parsed.success) { + return reply.status(400).send({ error: parsed.error.issues[0].message || 'Invalid parameters' }) + } + const network: NetworkName = req.network ?? parsed.data.network ?? activeNetwork + const rows = await getLatestLendingRates(network, { + asset, + poolId: parsed.data.poolId, + limit: parsed.data.limit, + }) + + return { + asset, + network, + rates: rows.map(r => ({ + id: r.id, + venue: r.source ?? 'blend', + source: r.source ?? 'blend', + network: r.network, + poolId: r.pool_id, + asset: r.asset, + assetAddress: r.asset_address, + supplyApy: Number(r.supply_apy), + borrowApy: Number(r.borrow_apy), + supplyApr: r.supply_apr !== null ? Number(r.supply_apr) : null, + borrowApr: r.borrow_apr !== null ? Number(r.borrow_apr) : null, + utilization: Number(r.utilization), + totalSupply: r.total_supply !== null ? Number(r.total_supply) : null, + totalBorrowed: r.total_borrowed !== null ? Number(r.total_borrowed) : null, + reserves: r.reserves !== null ? Number(r.reserves) : null, + ledger: r.ledger, + timestamp: r.timestamp instanceof Date ? r.timestamp.toISOString() : r.timestamp, + })), + } + }) +}