feat: added packet count metrics

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

@@ -200,8 +200,9 @@ export async function createTransfer(
   deps: ServiceDependencies,
   args: TransferOptions
 ): Promise<Transaction | TransferError> {
-  return createAccountToAccountTransfer(deps, {
+  return createAccountToAccountTransfer({
     transferArgs: args,
+    withdrawalThrottleDelay: deps.withdrawalThrottleDelay,
     voidTransfers: async (transferRefs) => voidTransfers(deps, transferRefs),
     postTransfers: async (transferRefs) => postTransfers(deps, transferRefs),
     getAccountReceived: async (accountRef) =>

packages/backend/src/accounting/service.ts

@@ -1,5 +1,4 @@
 import { TransactionOrKnex } from 'objection'
-import { BaseService } from '../shared/baseService'
 import { TransferError, isTransferError } from './errors'
 
 export enum LiquidityAccountType {
@@ -10,17 +9,13 @@ export enum LiquidityAccountType {
   WEB_MONETIZATION = 'WEB_MONETIZATION'
 }
 
-export interface LiquidityAccountAsset {
-  id: string
-  code?: string
-  scale?: number
-  ledger: number
-  onDebit?: (options: OnDebitOptions) => Promise<LiquidityAccount>
-}
-
 export interface LiquidityAccount {
   id: string
-  asset: LiquidityAccountAsset
+  asset: {
+    id: string
+    ledger: number
+    onDebit?: (options: OnDebitOptions) => Promise<LiquidityAccount>
+  }
   onCredit?: (options: OnCreditOptions) => Promise<LiquidityAccount>
   onDebit?: (options: OnDebitOptions) => Promise<LiquidityAccount>
 }
@@ -90,10 +85,6 @@ export interface TransferToCreate {
   ledger: number
 }
 
-export interface BaseAccountingServiceDependencies extends BaseService {
-  withdrawalThrottleDelay?: number
-}
-
 interface CreateAccountToAccountTransferArgs {
   transferArgs: TransferOptions
   voidTransfers(transferIds: string[]): Promise<void | TransferError>
@@ -103,10 +94,10 @@ interface CreateAccountToAccountTransferArgs {
   createPendingTransfers(
     transfers: TransferToCreate[]
   ): Promise<string[] | TransferError>
+  withdrawalThrottleDelay?: number
 }
 
 export async function createAccountToAccountTransfer(
-  deps: BaseAccountingServiceDependencies,
   args: CreateAccountToAccountTransferArgs
 ): Promise<Transaction | TransferError> {
   const {
@@ -115,11 +106,10 @@ export async function createAccountToAccountTransfer(
     createPendingTransfers,
     getAccountReceived,
     getAccountBalance,
+    withdrawalThrottleDelay,
     transferArgs
   } = args
 
-  const { withdrawalThrottleDelay } = deps
-
   const { sourceAccount, destinationAccount, sourceAmount, destinationAmount } =
     transferArgs
 

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

@@ -215,8 +215,9 @@ export async function createTransfer(
   deps: ServiceDependencies,
   args: TransferOptions
 ): Promise<Transaction | TransferError> {
-  return createAccountToAccountTransfer(deps, {
+  return createAccountToAccountTransfer({
     transferArgs: args,
+    withdrawalThrottleDelay: deps.withdrawalThrottleDelay,
     voidTransfers: async (transferIds) => {
       const error = await createTransfers(
         deps,

packages/backend/src/index.ts

@@ -287,6 +287,9 @@ export function initIocContainer(
       knex: await deps.use('knex'),
       accountingService: await deps.use('accountingService'),
       walletAddressService: await deps.use('walletAddressService'),
+      telemetry: config.enableTelemetry
+        ? await deps.use('telemetry')
+        : undefined,
       config: await deps.use('config')
     })
   })

packages/backend/src/open_payments/payment/incoming/service.ts

@@ -17,6 +17,7 @@ import {
 import { Amount } from '../../amount'
 import { IncomingPaymentError } from './errors'
 import { IAppConfig } from '../../../config/app'
+import { TelemetryService } from '../../../telemetry/service'
 
 export const POSITIVE_SLIPPAGE = BigInt(1)
 // First retry waits 10 seconds
@@ -47,6 +48,7 @@ export interface ServiceDependencies extends BaseService {
   accountingService: AccountingService
   walletAddressService: WalletAddressService
   config: IAppConfig
+  telemetry?: TelemetryService
 }
 
 export async function createIncomingPaymentService(
@@ -220,6 +222,11 @@ async function handleDeactivated(
       )
       await incomingPayment.$query(deps.knex).patch({ processAt: null })
       return
+    } else if (deps.telemetry) {
+      // TODO: tests
+      deps.telemetry.incrementCounter('transactions_count_incoming', 1, {
+        description: 'Count of funded incoming transactions'
+      })
     }
 
     const type =

packages/backend/src/open_payments/payment/outgoing/lifecycle.ts

@@ -85,8 +85,8 @@ export async function handleSending(
   const payEndTime = Date.now()
 
   if (deps.telemetry) {
-    deps.telemetry.incrementCounter('transactions_total', 1, {
-      description: 'Count of funded transactions'
+    deps.telemetry.incrementCounter('transactions_count_outgoing', 1, {
+      description: 'Count of funded outgoing transactions'
     })
 
     const payDuration = payEndTime - payStartTime

packages/backend/src/payment-method/ilp/connector/core/middleware/ilp-packet.ts

@@ -17,6 +17,7 @@ import {
 import { Readable } from 'stream'
 import { HttpContext, HttpMiddleware, ILPMiddleware } from '../rafiki'
 import getRawBody from 'raw-body'
+import { ValueType } from '@opentelemetry/api'
 
 export const CONTENT_TYPE = 'application/octet-stream'
 
@@ -234,5 +235,34 @@ export function createIlpPacketMiddleware(
     ctx.assert(!ctx.body, 500, 'response body already set')
     ctx.assert(response.rawReply, 500, 'ilp reply not set')
     ctx.body = response.rawReply
+
+    if (ctx.services.telemetry && Number(prepare.amount)) {
+      ctx.services.telemetry?.incrementCounter('packet_count_prepare', 1, {
+        description: 'Count of incoming prepare packets'
+      })
+      if (response.fulfill) {
+        const { code, scale } = ctx.state.incomingAccount.asset // Need to type these
+        const value = BigInt(prepare.amount)
+        await ctx.services.telemetry.incrementCounterWithTransactionAmount(
+          'transactions_amount',
+          {
+            value,
+            code,
+            scale
+          },
+          {
+            description: 'Amount sent through the network',
+            valueType: ValueType.DOUBLE
+          }
+        )
+        ctx.services.telemetry?.incrementCounter('packet_count_fulfill', 1, {
+          description: 'Count of outgoing fulfill packets'
+        })
+      } else if (response.reject) {
+        ctx.services.telemetry?.incrementCounter('packet_count_reject', 1, {
+          description: 'Count of outgoing reject packets'
+        })
+      }
+    }
   }
 }

packages/backend/src/payment-method/ilp/connector/core/test/middleware/ilp-packet-middleware.test.ts

@@ -1,3 +1,4 @@
+// import assert from 'assert'
 import { Readable } from 'stream'
 import {
   serializeIlpPrepare,
@@ -11,20 +12,46 @@ import {
   createIlpPacketMiddleware,
   IlpResponse,
   ZeroCopyIlpPrepare
-} from '../../middleware/ilp-packet'
+  // OutgoingAccount
+} from '../../'
 import {
   IlpPrepareFactory,
   IlpFulfillFactory,
-  IlpRejectFactory
+  IlpRejectFactory,
+  // IncomingAccountFactory,
+  RafikiServicesFactory
 } from '../../factories'
 import { HttpContext } from '../../rafiki'
 
 describe('ILP Packet Middleware', () => {
   test('sets up request and response', async () => {
+    // const incomingAccount = IncomingAccountFactory.build({ id: 'alice' })
+    // assert.ok(incomingAccount.id)
+    const services = RafikiServicesFactory.build({})
+
     const options: MockIncomingMessageOptions = {
       headers: { 'content-type': 'application/octet-stream' }
     }
-    const ctx = createContext<unknown, HttpContext>({ req: options })
+
+    const ctx = createContext<unknown, HttpContext>({
+      req: options,
+      services: { ...services, telemetry: undefined }
+      // accounts: {
+      //   incoming: incomingAccount,
+      //   outgoing: { asset: { code: 'USD', scale: 2 } } as OutgoingAccount
+      // },
+      // state: {
+      //   unfulfillable: false,
+      //   incomingAccount: {
+      //     quote: 'exists',
+      //     asset: {
+      //       code: 'USD',
+      //       scale: 2
+      //     }
+      //   }
+      // }
+    })
+
     const prepare = IlpPrepareFactory.build()
     const getRawBody = async (_req: Readable) => serializeIlpPrepare(prepare)
     const rawReply = serializeIlpFulfill(IlpFulfillFactory.build())