feat(backend): also publish webhooks to operators if primary recipient is tenant

packages/backend/migrations/20250409175316_create_webhooks_table.js

@@ -0,0 +1,66 @@
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = function (knex) {
+  return knex.schema
+    .createTable('webhooks', function (table) {
+      table.uuid('id').notNullable().primary()
+      table
+        .uuid('eventId')
+        .notNullable()
+        .references('webhookEvents.id')
+        .onDelete('CASCADE')
+        .index()
+      table
+        .uuid('recipientTenantId')
+        .notNullable()
+        .references('tenants.id')
+        .onDelete('CASCADE')
+        .index()
+
+      table.integer('attempts').notNullable().defaultTo(0)
+      table.integer('statusCode').nullable()
+
+      table.timestamp('processAt').nullable().defaultTo(knex.fn.now())
+
+      table.timestamp('createdAt').defaultTo(knex.fn.now())
+      table.timestamp('updatedAt').defaultTo(knex.fn.now())
+
+      table.index('processAt')
+    })
+    .then(() => {
+      return knex.raw(
+        `INSERT INTO "webhooks" (id, "eventId", "recipientTenantId", attempts, "statusCode", "processAt") select gen_random_uuid(), id as "eventId", "tenantId" as "recipientTenantId", attempts, "statusCode", "processAt" from "webhookEvents"`
+      )
+    })
+    .then(() => {
+      return knex.schema.alterTable('webhookEvents', (table) => {
+        table.dropColumn('attempts')
+        table.dropColumn('statusCode')
+        table.dropColumn('processAt')
+      })
+    })
+}
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = function (knex) {
+  return knex.schema
+    .alterTable('webhookEvents', function (table) {
+      table.integer('attempts').notNullable().defaultTo(0)
+      table.integer('statusCode').nullable()
+      table.timestamp('processAt').nullable().defaultTo(knex.fn.now())
+      table.index('processAt')
+    })
+    .then(() => {
+      return knex.raw(
+        `UPDATE "webhookEvents" SET "attempts" = (SELECT "attempts" from "webhooks" where "recipientTenantId" = "webhookEvents"."tenantId" AND "eventId" = "webhookEvents"."id"), "statusCode" = (SELECT "statusCode" from "webhooks" where "recipientTenantId" = "webhookEvents"."tenantId" AND "eventId" = "webhookEvents"."id"), "processAt" = (SELECT "processAt" from "webhooks" where "recipientTenantId" = "webhookEvents"."tenantId" AND "eventId" = "webhookEvents"."id")`
+      )
+    })
+    .then(() => {
+      return knex.schema.dropTableIfExists('webhooks')
+    })
+}

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'),
       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

@@ -27,6 +27,7 @@ describe('Ledger Account', (): void => {
     deps = initIocContainer({ ...Config, useTigerBeetle: false })
     appContainer = await createTestApp(deps)
     serviceDeps = {
+      config: await deps.use('config'),
       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

@@ -36,6 +36,7 @@ describe('Ledger Transfer', (): void => {
     deps = initIocContainer({ ...Config, useTigerBeetle: false })
     appContainer = await createTestApp(deps)
     serviceDeps = {
+      config: await deps.use('config'),
       logger: await deps.use('logger'),
       knex: await deps.use('knex'),
       telemetry: await deps.use('telemetry')

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

@@ -35,9 +35,11 @@ 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
+  config: IAppConfig
   telemetry: TelemetryService
   withdrawalThrottleDelay?: number
 }

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',
@@ -27,14 +28,20 @@ export interface LiquidityAccountAsset {
   code?: string
   scale?: number
   ledger: number
-  onDebit?: (options: OnDebitOptions) => Promise<LiquidityAccount>
+  onDebit?: (
+    options: OnDebitOptions,
+    config: IAppConfig
+  ) => Promise<LiquidityAccount>
 }
 
 export interface LiquidityAccount {
   id: string
   asset: LiquidityAccountAsset
   onCredit?: (options: OnCreditOptions) => Promise<LiquidityAccount>
-  onDebit?: (options: OnDebitOptions) => Promise<LiquidityAccount>
+  onDebit?: (
+    options: OnDebitOptions,
+    config: IAppConfig
+  ) => Promise<LiquidityAccount>
 }
 
 export interface OnCreditOptions {
@@ -133,6 +140,7 @@ export interface TransferToCreate {
 }
 
 export interface BaseAccountingServiceDependencies extends BaseService {
+  config: IAppConfig
   withdrawalThrottleDelay?: number
 }
 
@@ -195,15 +203,19 @@ export async function createAccountToAccountTransfer(
   }
 
   const onDebit = async (
-    account: LiquidityAccount | LiquidityAccount['asset']
+    account: LiquidityAccount | LiquidityAccount['asset'],
+    config: IAppConfig
   ) => {
     if (account.onDebit) {
       const balance = await getAccountBalance(account.id)
       if (balance === undefined) throw new Error('undefined account balance')
 
-      await account.onDebit({
-        balance
-      })
+      await account.onDebit(
+        {
+          balance
+        },
+        config
+      )
     }
   }
 
@@ -213,8 +225,8 @@ export async function createAccountToAccountTransfer(
       if (error) return error
 
       await Promise.all([
-        onDebit(sourceAccount),
-        onDebit(destinationAccount.asset)
+        onDebit(sourceAccount, deps.config),
+        onDebit(destinationAccount.asset, deps.config)
       ])
 
       if (destinationAccount.onCredit) {

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,
@@ -68,6 +69,7 @@ export const convertToTigerBeetleTransferCode: {
 }
 
 export interface ServiceDependencies extends BaseService {
+  config: IAppConfig
   tigerBeetle: Client
   telemetry: TelemetryService
   withdrawalThrottleDelay?: number

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

@@ -1,7 +1,7 @@
 import { Knex } from 'knex'
 
 import { AssetService } from './service'
-import { Config } from '../config/app'
+import { Config, IAppConfig } from '../config/app'
 import { createTestApp, TestContainer } from '../tests/app'
 import { IocContract } from '@adonisjs/fold'
 import { initIocContainer } from '../'
@@ -10,6 +10,7 @@ import { randomAsset } from '../tests/asset'
 import { truncateTables } from '../tests/tableManager'
 import { Asset, AssetEvent, AssetEventError, AssetEventType } from './model'
 import { isAssetError } from './errors'
+import { createTenant } from '../tests/tenant'
 
 describe('Models', (): void => {
   let deps: IocContract<AppServices>
@@ -35,7 +36,9 @@ describe('Models', (): void => {
   describe('Asset Model', (): void => {
     describe('onDebit', (): void => {
       let asset: Asset
+      let config: IAppConfig
       beforeEach(async (): Promise<void> => {
+        config = await deps.use('config')
         const options = {
           ...randomAsset(),
           tenantId: Config.operatorTenantId,
@@ -54,13 +57,13 @@ describe('Models', (): void => {
       `(
         'creates webhook event if balance=$balance <= liquidityThreshold',
         async ({ balance }): Promise<void> => {
-          await asset.onDebit({ balance })
+          await asset.onDebit({ balance }, config)
           const event = (
-            await AssetEvent.query(knex).where(
-              'type',
-              AssetEventType.LiquidityLow
-            )
+            await AssetEvent.query(knex)
+              .where('type', AssetEventType.LiquidityLow)
+              .withGraphFetched('webhooks')
           )[0]
+          expect(event.webhooks).toHaveLength(1)
           expect(event).toMatchObject({
             type: AssetEventType.LiquidityLow,
             data: {
@@ -72,16 +75,78 @@ describe('Models', (): void => {
               },
               liquidityThreshold: asset.liquidityThreshold?.toString(),
               balance: balance.toString()
-            }
+            },
+            webhooks: expect.arrayContaining([
+              expect.objectContaining({
+                id: expect.any(String),
+                eventId: event.id,
+                recipientTenantId: Config.operatorTenantId,
+                processAt: expect.any(Date),
+                attempts: 0
+              })
+            ])
           })
         }
       )
       test('does not create webhook event if balance > liquidityThreshold', async (): Promise<void> => {
-        await asset.onDebit({ balance: BigInt(110) })
+        await asset.onDebit({ balance: BigInt(110) }, config)
         await expect(
           AssetEvent.query(knex).where('type', AssetEventType.LiquidityLow)
         ).resolves.toEqual([])
       })
+      test('creates corresponding webhook for operator if asset belongs to tenant', async (): Promise<void> => {
+        const tenant = await createTenant(deps)
+        const tenantAssetOptions = {
+          ...randomAsset(),
+          tenantId: tenant.id,
+          liquidityThreshold: BigInt(100)
+        }
+
+        let tenantAsset: Asset
+        const assetOrError = await assetService.create(tenantAssetOptions)
+        if (!isAssetError(assetOrError)) {
+          tenantAsset = assetOrError
+        } else {
+          throw assetOrError
+        }
+
+        await tenantAsset.onDebit({ balance: BigInt(50) }, config)
+        const event = (
+          await AssetEvent.query(knex)
+            .where('type', AssetEventType.LiquidityLow)
+            .withGraphFetched('webhooks')
+        )[0]
+        expect(event.webhooks).toHaveLength(2)
+        expect(event).toMatchObject({
+          type: AssetEventType.LiquidityLow,
+          data: {
+            id: tenantAsset.id,
+            asset: {
+              id: tenantAsset.id,
+              code: tenantAsset.code,
+              scale: tenantAsset.scale
+            },
+            liquidityThreshold: tenantAsset.liquidityThreshold?.toString(),
+            balance: '50'
+          },
+          webhooks: expect.arrayContaining([
+            expect.objectContaining({
+              id: expect.any(String),
+              eventId: event.id,
+              recipientTenantId: Config.operatorTenantId,
+              processAt: expect.any(Date),
+              attempts: 0
+            }),
+            expect.objectContaining({
+              id: expect.any(String),
+              eventId: event.id,
+              recipientTenantId: tenant.id,
+              processAt: expect.any(Date),
+              attempts: 0
+            })
+          ])
+        })
+      })
     })
   })
 

packages/backend/src/asset/model.ts

@@ -1,7 +1,8 @@
 import { QueryContext } from 'objection'
 import { LiquidityAccount, OnDebitOptions } from '../accounting/service'
 import { BaseModel } from '../shared/baseModel'
-import { WebhookEvent } from '../webhook/model'
+import { WebhookEvent } from '../webhook/event/model'
+import { IAppConfig } from '../config/app'
 
 export class Asset extends BaseModel implements LiquidityAccount {
   public static get tableName(): string {
@@ -29,10 +30,17 @@ export class Asset extends BaseModel implements LiquidityAccount {
     }
   }
 
-  public async onDebit({ balance }: OnDebitOptions): Promise<Asset> {
+  public async onDebit(
+    { balance }: OnDebitOptions,
+    config: IAppConfig
+  ): Promise<Asset> {
     if (this.liquidityThreshold !== null) {
       if (balance <= this.liquidityThreshold) {
-        await AssetEvent.query().insert({
+        const webhooks = [{ recipientTenantId: this.tenantId }]
+        if (this.tenantId !== config.operatorTenantId) {
+          webhooks.push({ recipientTenantId: config.operatorTenantId })
+        }
+        await AssetEvent.query().insertGraph({
           assetId: this.id,
           type: AssetEventType.LiquidityLow,
           data: {
@@ -45,7 +53,8 @@ export class Asset extends BaseModel implements LiquidityAccount {
             liquidityThreshold: this.liquidityThreshold,
             balance
           },
-          tenantId: this.tenantId
+          tenantId: this.tenantId,
+          webhooks
         })
       }
     }