feat(backend): tenanted webhooks

packages/backend/migrations/20250218191304_add_tenant_to_webhooks.js

@@ -0,0 +1,30 @@
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = function (knex) {
+  return knex.schema
+    .alterTable('webhookEvents', (table) => {
+      table.uuid('tenantId').references('tenants.id').index()
+    })
+    .then(() => {
+      return knex.raw(
+        `UPDATE "webhookEvents" SET "tenantId" = (SELECT id from "tenants" LIMIT 1)`
+      )
+    })
+    .then(() => {
+      return knex.schema.alterTable('webhookEvents', (table) => {
+        table.uuid('tenantId').notNullable().alter()
+      })
+    })
+}
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = function (knex) {
+  return knex.schema.alterTable('webhookEvents', function (table) {
+    table.dropColumn('tenantId')
+  })
+}

packages/backend/src/accounting/psql/balance.test.ts

@@ -26,6 +26,7 @@ describe('Balances', (): void => {
     deps = initIocContainer({ ...Config, useTigerBeetle: false })
     appContainer = await createTestApp(deps)
     serviceDeps = {
+      config: await deps.use('config'), // TODO: remove once tenanted peers are added
       logger: await deps.use('logger'),
       knex: await deps.use('knex'),
       telemetry: await deps.use('telemetry')

packages/backend/src/accounting/psql/ledger-account/index.test.ts

@@ -24,6 +24,7 @@ describe('Ledger Account', (): void => {
     const deps = initIocContainer({ ...Config, useTigerBeetle: false })
     appContainer = await createTestApp(deps)
     serviceDeps = {
+      config: await deps.use('config'), // TODO: remove once tenanted peers are in
       logger: await deps.use('logger'),
       knex: await deps.use('knex'),
       telemetry: await deps.use('telemetry')

packages/backend/src/accounting/psql/ledger-transfer/index.test.ts

@@ -33,6 +33,7 @@ describe('Ledger Transfer', (): void => {
     const deps = initIocContainer({ ...Config, useTigerBeetle: false })
     appContainer = await createTestApp(deps)
     serviceDeps = {
+      config: await deps.use('config'), // TODO: remove once tenanted peers are in
       logger: await deps.use('logger'),
       knex: await deps.use('knex'),
       telemetry: await deps.use('telemetry')

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

@@ -35,11 +35,13 @@ import {
 } from './ledger-transfer'
 import { LedgerTransfer, LedgerTransferType } from './ledger-transfer/model'
 import { TelemetryService } from '../../telemetry/service'
+import { IAppConfig } from '../../config/app'
 
 export interface ServiceDependencies extends BaseService {
   knex: TransactionOrKnex
   telemetry: TelemetryService
   withdrawalThrottleDelay?: number
+  config: IAppConfig // TODO: remove when peers are tenanted
 }
 
 export function createAccountingService(

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 { IAppConfig } from '../config/app'
 
 export enum LiquidityAccountType {
   ASSET = 'ASSET',
@@ -44,6 +45,7 @@ export interface OnCreditOptions {
 
 export interface OnDebitOptions {
   balance: bigint
+  tenantId: string
 }
 
 export interface BaseTransfer {
@@ -134,6 +136,7 @@ export interface TransferToCreate {
 
 export interface BaseAccountingServiceDependencies extends BaseService {
   withdrawalThrottleDelay?: number
+  config: IAppConfig // TODO: Remove when peers are tenanted
 }
 
 interface CreateAccountToAccountTransferArgs {
@@ -201,8 +204,10 @@ export async function createAccountToAccountTransfer(
       const balance = await getAccountBalance(account.id)
       if (balance === undefined) throw new Error('undefined account balance')
 
+      const { config } = deps
       await account.onDebit({
-        balance
+        balance,
+        tenantId: config.operatorTenantId
       })
     }
   }

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

@@ -32,6 +32,7 @@ import {
 } from './transfers'
 import { toTigerBeetleId } from './utils'
 import { TelemetryService } from '../../telemetry/service'
+import { IAppConfig } from '../../config/app'
 
 export enum TigerBeetleAccountCode {
   LIQUIDITY_WEB_MONETIZATION = 1,
@@ -71,6 +72,7 @@ export interface ServiceDependencies extends BaseService {
   tigerBeetle: Client
   telemetry: TelemetryService
   withdrawalThrottleDelay?: number
+  config: IAppConfig // TODO: remove when peers are tenanted
 }
 
 export function createAccountingService(

packages/backend/src/asset/model.test.ts

@@ -54,7 +54,7 @@ describe('Models', (): void => {
       `(
         'creates webhook event if balance=$balance <= liquidityThreshold',
         async ({ balance }): Promise<void> => {
-          await asset.onDebit({ balance })
+          await asset.onDebit({ balance, tenantId: asset.tenantId }) // TODO: remove once tenanted peers are in
           const event = (
             await AssetEvent.query(knex).where(
               'type',
@@ -77,7 +77,7 @@ describe('Models', (): void => {
         }
       )
       test('does not create webhook event if balance > liquidityThreshold', async (): Promise<void> => {
-        await asset.onDebit({ balance: BigInt(110) })
+        await asset.onDebit({ balance: BigInt(110), tenantId: asset.tenantId }) // TODO: remove once tenanted peers are in
         await expect(
           AssetEvent.query(knex).where('type', AssetEventType.LiquidityLow)
         ).resolves.toEqual([])

packages/backend/src/asset/model.ts

@@ -44,7 +44,8 @@ export class Asset extends BaseModel implements LiquidityAccount {
             },
             liquidityThreshold: this.liquidityThreshold,
             balance
-          }
+          },
+          tenantId: this.tenantId
         })
       }
     }

packages/backend/src/graphql/resolvers/liquidity.test.ts

@@ -1796,7 +1796,8 @@ describe('Liquidity Resolvers', (): void => {
               data: payment.toData({
                 amountSent: BigInt(0),
                 balance: BigInt(0)
-              })
+              }),
+              tenantId: Config.operatorTenantId
             })
           })
 
@@ -1956,6 +1957,7 @@ describe('Liquidity Resolvers', (): void => {
       incomingPaymentId?: string
       outgoingPaymentId?: string
       walletAddressId?: string
+      tenantId?: string
     }
 
     const isIncomingPaymentEventType = (
@@ -2011,7 +2013,8 @@ describe('Liquidity Resolvers', (): void => {
                 accountId: liquidityAccount.id,
                 assetId: liquidityAccount.asset.id,
                 amount
-              }
+              },
+              tenantId: Config.operatorTenantId
             }
 
             if (resourceId) {
@@ -2230,7 +2233,8 @@ describe('Liquidity Resolvers', (): void => {
               accountId: incomingPayment.id,
               assetId: incomingPayment.asset.id,
               amount
-            }
+            },
+            tenantId: Config.operatorTenantId
           })
 
           const response = await appContainer.apolloClient
@@ -2278,7 +2282,8 @@ describe('Liquidity Resolvers', (): void => {
               accountId: incomingPayment.id,
               assetId: incomingPayment.asset.id,
               amount
-            }
+            },
+            tenantId: Config.operatorTenantId
           })
           let error
           try {
@@ -2376,7 +2381,8 @@ describe('Liquidity Resolvers', (): void => {
               accountId: incomingPayment.id,
               assetId: incomingPayment.asset.id,
               amount
-            }
+            },
+            tenantId: Config.operatorTenantId
           })
           await expect(
             accountingService.createWithdrawal({
@@ -2463,7 +2469,8 @@ describe('Liquidity Resolvers', (): void => {
               accountId: outgoingPayment.id,
               assetId: outgoingPayment.asset.id,
               amount
-            }
+            },
+            tenantId: Config.operatorTenantId
           })
 
           const response = await appContainer.apolloClient
@@ -2604,7 +2611,8 @@ describe('Liquidity Resolvers', (): void => {
               accountId: outgoingPayment.id,
               assetId: outgoingPayment.asset.id,
               amount
-            }
+            },
+            tenantId: Config.operatorTenantId
           })
           await expect(
             accountingService.createWithdrawal({
@@ -2673,7 +2681,8 @@ describe('Liquidity Resolvers', (): void => {
               data: outgoingPayment.toData({
                 amountSent: BigInt(0),
                 balance: BigInt(0)
-              })
+              }),
+              tenantId: Config.operatorTenantId
             })
           })
 

packages/backend/src/index.ts

@@ -357,6 +357,7 @@ export function initIocContainer(
       return createTigerbeetleAccountingService({
         logger,
         knex,
+        config,
         tigerBeetle,
         withdrawalThrottleDelay: config.withdrawalThrottleDelay,
         telemetry
@@ -366,6 +367,7 @@ export function initIocContainer(
     return createPsqlAccountingService({
       logger,
       knex,
+      config,
       withdrawalThrottleDelay: config.withdrawalThrottleDelay,
       telemetry
     })
@@ -395,6 +397,7 @@ export function initIocContainer(
   })
   container.singleton('webhookService', async (deps) => {
     return createWebhookService({
+      tenantSettingService: await deps.use('tenantSettingService'),
       config: await deps.use('config'),
       knex: await deps.use('knex'),
       logger: await deps.use('logger')

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

@@ -324,7 +324,8 @@ describe('Incoming Payment Service', (): void => {
       const incomingPayment = await incomingPaymentService.create({
         walletAddressId,
         ...options,
-        incomingAmount: options.incomingAmount ? amount : undefined
+        incomingAmount: options.incomingAmount ? amount : undefined,
+        tenantId: Config.operatorTenantId
       })
       assert.ok(!isIncomingPaymentError(incomingPayment))
       expect(incomingPayment).toMatchObject({

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

@@ -199,7 +199,8 @@ async function createIncomingPayment(
   await IncomingPaymentEvent.query(trx || deps.knex).insert({
     incomingPaymentId: incomingPayment.id,
     type: IncomingPaymentEventType.IncomingPaymentCreated,
-    data: incomingPayment.toData(0n)
+    data: incomingPayment.toData(0n),
+    tenantId: incomingPayment.tenantId
   })
 
   incomingPayment = await addReceivedAmount(deps, incomingPayment, BigInt(0))
@@ -364,7 +365,8 @@ async function handleDeactivated(
         accountId: incomingPayment.id,
         assetId: incomingPayment.assetId,
         amount: amountReceived
-      }
+      },
+      tenantId: incomingPayment.tenantId
     })
 
     await incomingPayment.$query(deps.knex).patch({

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

@@ -225,7 +225,8 @@ export async function sendWebhookEvent(
     outgoingPaymentId: payment.id,
     type,
     data: payment.toData({ amountSent, balance }),
-    withdrawal
+    withdrawal,
+    tenantId: payment.tenantId
   })
   stopTimer()
 }

packages/backend/src/open_payments/wallet_address/service.ts

@@ -313,7 +313,8 @@ async function getOrPollByUrl(
     type: WalletAddressEventType.WalletAddressNotFound,
     data: {
       walletAddressUrl: url
-    }
+    },
+    tenantId: deps.config.operatorTenantId
   })
 
   deps.logger.debug(
@@ -458,7 +459,8 @@ async function createWithdrawalEvent(
       accountId: walletAddress.id,
       assetId: walletAddress.assetId,
       amount
-    }
+    },
+    tenantId: walletAddress.tenantId
   })
 
   await walletAddress.$query(deps.knex).patch({

packages/backend/src/payment-method/ilp/peer/model.test.ts

@@ -73,7 +73,7 @@ describe('Models', (): void => {
       `(
         'creates webhook event if balance=$balance <= liquidityThreshold',
         async ({ balance }): Promise<void> => {
-          await peer.onDebit({ balance })
+          await peer.onDebit({ balance, tenantId: Config.operatorTenantId })
           const event = (
             await PeerEvent.query(knex).where(
               'type',
@@ -91,12 +91,16 @@ describe('Models', (): void => {
               },
               liquidityThreshold: peer.liquidityThreshold?.toString(),
               balance: balance.toString()
-            }
+            },
+            tenantId: Config.operatorTenantId
           })
         }
       )
       test('does not create webhook event if balance > liquidityThreshold', async (): Promise<void> => {
-        await peer.onDebit({ balance: BigInt(110) })
+        await peer.onDebit({
+          balance: BigInt(110),
+          tenantId: Config.operatorTenantId
+        })
         await expect(
           PeerEvent.query(knex).where('type', PeerEventType.LiquidityLow)
         ).resolves.toEqual([])

packages/backend/src/payment-method/ilp/peer/model.ts

@@ -53,7 +53,7 @@ export class Peer
 
   public name?: string
 
-  public async onDebit({ balance }: OnDebitOptions): Promise<Peer> {
+  public async onDebit({ balance, tenantId }: OnDebitOptions): Promise<Peer> {
     if (this.liquidityThreshold !== null) {
       if (balance <= this.liquidityThreshold) {
         await PeerEvent.query().insert({
@@ -68,7 +68,8 @@ export class Peer
             },
             liquidityThreshold: this.liquidityThreshold,
             balance
-          }
+          },
+          tenantId // TODO: update when peers are tenanted
         })
       }
     }