feat(workweek): asset caching.

localenv/telemetry/grafana/provisioning/dashboards/example.json

@@ -90,9 +90,7 @@
       "id": 12,
       "options": {
         "legend": {
-          "calcs": [
-            "p95"
-          ],
+          "calcs": ["p95"],
           "displayMode": "table",
           "placement": "right",
           "showLegend": true
@@ -201,9 +199,7 @@
       "id": 9,
       "options": {
         "legend": {
-          "calcs": [
-            "p95"
-          ],
+          "calcs": ["p95"],
           "displayMode": "table",
           "placement": "right",
           "showLegend": true
@@ -365,9 +361,7 @@
       "id": 7,
       "options": {
         "legend": {
-          "calcs": [
-            "p95"
-          ],
+          "calcs": ["p95"],
           "displayMode": "table",
           "placement": "right",
           "showLegend": true
@@ -624,9 +618,7 @@
       "id": 10,
       "options": {
         "legend": {
-          "calcs": [
-            "p95"
-          ],
+          "calcs": ["p95"],
           "displayMode": "table",
           "placement": "right",
           "showLegend": true
@@ -776,9 +768,7 @@
         "namePlacement": "auto",
         "orientation": "horizontal",
         "reduceOptions": {
-          "calcs": [
-            "lastNotNull"
-          ],
+          "calcs": ["lastNotNull"],
           "fields": "",
           "values": false
         },
@@ -850,9 +840,7 @@
         "footer": {
           "countRows": false,
           "fields": "",
-          "reducer": [
-            "sum"
-          ],
+          "reducer": ["sum"],
           "show": false
         },
         "showHeader": true,
@@ -881,9 +869,7 @@
               "operator": "=",
               "scope": "resource",
               "tag": "service.name",
-              "value": [
-                "RAFIKI_NETWORK"
-              ],
+              "value": ["RAFIKI_NETWORK"],
               "valueType": "string"
             },
             {
@@ -1150,9 +1136,7 @@
         "orientation": "auto",
         "percentChangeColorMode": "standard",
         "reduceOptions": {
-          "calcs": [
-            "lastNotNull"
-          ],
+          "calcs": ["lastNotNull"],
           "fields": "",
           "values": false
         },
@@ -1247,9 +1231,7 @@
         "orientation": "auto",
         "percentChangeColorMode": "standard",
         "reduceOptions": {
-          "calcs": [
-            "lastNotNull"
-          ],
+          "calcs": ["lastNotNull"],
           "fields": "",
           "values": false
         },
@@ -1330,4 +1312,4 @@
   "uid": "fdr58stwkr6yof",
   "version": 2,
   "weekStart": ""
-}
+}

packages/backend/src/accounting/psql/service.ts

@@ -35,8 +35,10 @@ import {
 } from './ledger-transfer'
 import { LedgerTransfer, LedgerTransferType } from './ledger-transfer/model'
 import { TelemetryService } from '../../telemetry/service'
+import { AssetService } from '../../asset/service'
 
 export interface ServiceDependencies extends BaseService {
+  fetchAssetService?: () => Promise<AssetService>
   knex: TransactionOrKnex
   withdrawalThrottleDelay?: number
   telemetry?: TelemetryService

packages/backend/src/accounting/service.ts

@@ -1,6 +1,7 @@
 import { TransactionOrKnex } from 'objection'
 import { BaseService } from '../shared/baseService'
 import { TransferError, isTransferError } from './errors'
+import { AssetService } from '../asset/service'
 
 export enum LiquidityAccountType {
   ASSET = 'ASSET',
@@ -40,6 +41,7 @@ export interface LiquidityAccount {
 export interface OnCreditOptions {
   totalReceived: bigint
   withdrawalThrottleDelay?: number
+  fetchAssetService?: () => Promise<AssetService>
 }
 
 export interface OnDebitOptions {
@@ -133,6 +135,7 @@ export interface TransferToCreate {
 }
 
 export interface BaseAccountingServiceDependencies extends BaseService {
+  fetchAssetService?: () => Promise<AssetService>
   withdrawalThrottleDelay?: number
 }
 
@@ -160,7 +163,7 @@ export async function createAccountToAccountTransfer(
     getAccountBalance
   } = args
 
-  const { withdrawalThrottleDelay } = deps
+  const { withdrawalThrottleDelay, fetchAssetService } = deps
 
   const { sourceAccount, destinationAccount, sourceAmount, destinationAmount } =
     transferArgs
@@ -226,7 +229,8 @@ export async function createAccountToAccountTransfer(
 
         await destinationAccount.onCredit({
           totalReceived,
-          withdrawalThrottleDelay
+          withdrawalThrottleDelay,
+          fetchAssetService
         })
       }
     },

packages/backend/src/accounting/tigerbeetle/service.ts

@@ -32,6 +32,7 @@ import {
 } from './transfers'
 import { toTigerBeetleId } from './utils'
 import { TelemetryService } from '../../telemetry/service'
+import { AssetService } from '../../asset/service'
 
 export enum TigerBeetleAccountCode {
   LIQUIDITY_WEB_MONETIZATION = 1,
@@ -68,6 +69,7 @@ export const convertToTigerBeetleTransferCode: {
 }
 
 export interface ServiceDependencies extends BaseService {
+  fetchAssetService?: () => Promise<AssetService>
   tigerBeetle: Client
   withdrawalThrottleDelay?: number
   telemetry?: TelemetryService

packages/backend/src/asset/service.ts

@@ -7,6 +7,8 @@ import { BaseService } from '../shared/baseService'
 import { AccountingService, LiquidityAccountType } from '../accounting/service'
 import { WalletAddress } from '../open_payments/wallet_address/model'
 import { Peer } from '../payment-method/ilp/peer/model'
+import { Quote } from '../open_payments/quote/model'
+import { IncomingPayment } from '../open_payments/payment/incoming/model'
 
 export interface AssetOptions {
   code: string
@@ -28,30 +30,33 @@ export interface DeleteOptions {
   deletedAt: Date
 }
 
+export type ToSetOn = Quote | IncomingPayment | WalletAddress | Peer | undefined
+
 export interface AssetService {
   create(options: CreateOptions): Promise<Asset | AssetError>
   update(options: UpdateOptions): Promise<Asset | AssetError>
   delete(options: DeleteOptions): Promise<Asset | AssetError>
   get(id: string): Promise<void | Asset>
+  setOn(obj: ToSetOn): Promise<void | Asset>
   getPage(pagination?: Pagination, sortOrder?: SortOrder): Promise<Asset[]>
   getAll(): Promise<Asset[]>
 }
 
 interface ServiceDependencies extends BaseService {
-  accountingService: AccountingService,
+  accountingService: AccountingService
   assetCache: Map<string, Asset>
 }
 
 export async function createAssetService({
   logger,
   knex,
-  accountingService
+  accountingService,
+  assetCache
 }: ServiceDependencies): Promise<AssetService> {
   const log = logger.child({
     service: 'AssetService'
   })
 
-  const assetCache = new Map<string, Asset>()
   const deps: ServiceDependencies = {
     logger: log,
     knex,
@@ -64,6 +69,7 @@ export async function createAssetService({
     update: (options) => updateAsset(deps, options),
     delete: (options) => deleteAsset(deps, options),
     get: (id) => getAsset(deps, id),
+    setOn: (toSetOn) => setAssetOn(deps, toSetOn),
     getPage: (pagination?, sortOrder?) =>
       getAssetsPage(deps, pagination, sortOrder),
     getAll: () => getAll(deps)
@@ -182,7 +188,10 @@ async function getAsset(
   const inMem = deps.assetCache.get(id)
   if (inMem) return inMem
 
-  return Asset.query(deps.knex).whereNull('deletedAt').findById(id)
+  const asset = await Asset.query(deps.knex).whereNull('deletedAt').findById(id)
+  if (asset) deps.assetCache.set(asset.id, asset)
+
+  return asset
 }
 
 async function getAssetsPage(
@@ -198,3 +207,12 @@ async function getAssetsPage(
 async function getAll(deps: ServiceDependencies): Promise<Asset[]> {
   return Asset.query(deps.knex).whereNull('deletedAt')
 }
+
+async function setAssetOn(
+  deps: ServiceDependencies,
+  obj: ToSetOn
+): Promise<void> {
+  if (!obj) return
+  const asset = await getAsset(deps, obj.assetId)
+  if (asset) obj.asset = asset
+}

packages/backend/src/config/app.ts

@@ -122,16 +122,16 @@ export const Config = {
   quoteLifespan: envInt('QUOTE_LIFESPAN', 5 * 60_000), // milliseconds
 
   walletAddressWorkers: envInt('WALLET_ADDRESS_WORKERS', 1),
-  walletAddressWorkerIdle: envInt('WALLET_ADDRESS_WORKER_IDLE', 200), // milliseconds
+  walletAddressWorkerIdle: envInt('WALLET_ADDRESS_WORKER_IDLE', 3000), // milliseconds
 
   authServerGrantUrl: envString('AUTH_SERVER_GRANT_URL'),
   authServerIntrospectionUrl: envString('AUTH_SERVER_INTROSPECTION_URL'),
 
   outgoingPaymentWorkers: envInt('OUTGOING_PAYMENT_WORKERS', 1),
-  outgoingPaymentWorkerIdle: envInt('OUTGOING_PAYMENT_WORKER_IDLE', 200), // milliseconds
+  outgoingPaymentWorkerIdle: envInt('OUTGOING_PAYMENT_WORKER_IDLE', 3000), // milliseconds
 
   incomingPaymentWorkers: envInt('INCOMING_PAYMENT_WORKERS', 1),
-  incomingPaymentWorkerIdle: envInt('INCOMING_PAYMENT_WORKER_IDLE', 200), // milliseconds
+  incomingPaymentWorkerIdle: envInt('INCOMING_PAYMENT_WORKER_IDLE', 3000), // milliseconds
   pollIncomingPaymentCreatedWebhook: envBool(
     'POLL_INCOMING_PAYMENT_CREATED_WEBHOOK',
     false
@@ -146,7 +146,7 @@ export const Config = {
   ), // milliseconds
 
   webhookWorkers: envInt('WEBHOOK_WORKERS', 1),
-  webhookWorkerIdle: envInt('WEBHOOK_WORKER_IDLE', 200), // milliseconds
+  webhookWorkerIdle: envInt('WEBHOOK_WORKER_IDLE', 3000), // milliseconds
   webhookUrl: envString('WEBHOOK_URL'),
   webhookTimeout: envInt('WEBHOOK_TIMEOUT', 2000), // milliseconds
   webhookMaxRetry: envInt('WEBHOOK_MAX_RETRY', 10),

packages/backend/src/index.ts

@@ -56,6 +56,7 @@ import { createStreamCredentialsService } from './payment-method/ilp/stream-cred
 import { createRatesService } from './rates/service'
 import { TelemetryService, createTelemetryService } from './telemetry/service'
 import { createWebhookService } from './webhook/service'
+import { Asset } from './asset/model'
 
 BigInt.prototype.toJSON = function () {
   return this.toString()
@@ -202,9 +203,10 @@ export function initIocContainer(
     const logger = await deps.use('logger')
     const knex = await deps.use('knex')
     return await createAssetService({
-      logger: logger,
-      knex: knex,
-      accountingService: await deps.use('accountingService')
+      logger,
+      knex,
+      accountingService: await deps.use('accountingService'),
+      assetCache: new Map<string, Asset>()
     })
   })
 
@@ -232,15 +234,17 @@ export function initIocContainer(
         knex,
         tigerBeetle,
         withdrawalThrottleDelay: config.withdrawalThrottleDelay,
-        telemetry
+        telemetry,
+        fetchAssetService: () => Promise.resolve(deps.use('assetService'))
       })
     }
 
     return createPsqlAccountingService({
       logger,
       knex,
       withdrawalThrottleDelay: config.withdrawalThrottleDelay,
-      telemetry
+      telemetry,
+      fetchAssetService: () => Promise.resolve(deps.use('assetService'))
     })
   })
   container.singleton('peerService', async (deps) => {
@@ -280,7 +284,8 @@ export function initIocContainer(
       knex: await deps.use('knex'),
       logger: logger,
       accountingService: await deps.use('accountingService'),
-      webhookService: await deps.use('webhookService')
+      webhookService: await deps.use('webhookService'),
+      assetService: await deps.use('assetService')
     })
   })
   container.singleton('spspRoutes', async (deps) => {
@@ -298,7 +303,8 @@ export function initIocContainer(
       knex: await deps.use('knex'),
       accountingService: await deps.use('accountingService'),
       walletAddressService: await deps.use('walletAddressService'),
-      config: await deps.use('config')
+      config: await deps.use('config'),
+      assetService: await deps.use('assetService')
     })
   })
   container.singleton('remoteIncomingPaymentService', async (deps) => {
@@ -469,7 +475,8 @@ export function initIocContainer(
       ),
       telemetry: config.enableTelemetry
         ? await deps.use('telemetry')
-        : undefined
+        : undefined,
+      assetService: await deps.use('assetService')
     })
   })