feat(auth, backend): add migration files for updating Rafiki resources with multiple tenant support

localenv/cloud-nine-wallet/dbinit.sql

@@ -9,3 +9,7 @@ ALTER DATABASE cloud_nine_wallet_auth OWNER TO cloud_nine_wallet_auth;
 CREATE USER happy_life_bank_backend WITH PASSWORD 'happy_life_bank_backend';
 CREATE DATABASE happy_life_bank_backend;
 ALTER DATABASE happy_life_bank_backend OWNER TO happy_life_bank_backend;
+
+CREATE USER happy_life_bank_auth WITH PASSWORD 'happy_life_bank_auth';
+CREATE DATABASE happy_life_bank_auth;
+ALTER DATABASE happy_life_bank_auth OWNER TO happy_life_bank_auth;

packages/auth/migrations/20240827131401_create_tenants_tables.js

@@ -0,0 +1,22 @@
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = function (knex) {
+  return knex.schema.createTable('tenants', function (table) {
+    table.uuid('id').primary()
+    table.string('idpConsentEndpoint').notNullable()
+    table.string('idpSecret').notNullable()
+    table.timestamp('createdAt').defaultTo(knex.fn.now())
+    table.timestamp('updatedAt').defaultTo(knex.fn.now())
+    table.timestamp('deletedAt').nullable().defaultTo(null)
+  })
+}
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = function (knex) {
+  return knex.schema.dropTable('tenants')
+}

packages/auth/migrations/20240827131417_update_grants_with_tenantId.js

@@ -0,0 +1,21 @@
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = function (knex) {
+  return knex.schema.table('grants', function (table) {
+    table.uuid('tenantId').notNullable()
+    table.foreign('tenantId').references('id').inTable('tenants')
+  })
+}
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = function (knex) {
+  return knex.schema.table('grants', function (table) {
+    table.dropForeign(['tenantId'])
+    table.dropColumn('tenantId')
+  })
+}

packages/backend/migrations/20240827115808_create_tenants_tables.js

@@ -0,0 +1,21 @@
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = function (knex) {
+  return knex.schema.createTable('tenants', function (table) {
+    table.uuid('id').primary()
+    table.string('kratosIdentityId').notNullable()
+    table.timestamp('createdAt').defaultTo(knex.fn.now())
+    table.timestamp('updatedAt').defaultTo(knex.fn.now())
+    table.timestamp('deletedAt').nullable().defaultTo(null)
+  })
+}
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = function (knex) {
+  return knex.schema.dropTable('tenants')
+}

packages/backend/migrations/20240827115832_update_resources_with_tenantId.js

@@ -0,0 +1,55 @@
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = function (knex) {
+  return knex.schema
+    .table('quotes', function (table) {
+      table.uuid('tenantId').notNullable()
+      table.foreign('tenantId').references('id').inTable('tenants')
+    })
+    .table('incomingPayments', function (table) {
+      table.uuid('tenantId').notNullable()
+      table.foreign('tenantId').references('id').inTable('tenants')
+    })
+    .table('outgoingPayments', function (table) {
+      table.uuid('tenantId').notNullable()
+      table.foreign('tenantId').references('id').inTable('tenants')
+    })
+    .table('walletAddresses', function (table) {
+      table.uuid('tenantId').notNullable()
+      table.foreign('tenantId').references('id').inTable('tenants')
+    })
+    .table('grants', function (table) {
+      table.uuid('tenantId').notNullable()
+      table.foreign('tenantId').references('id').inTable('tenants')
+    })
+}
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = function (knex) {
+  return knex.schema
+    .table('quotes', function (table) {
+      table.dropForeign(['tenantId'])
+      table.dropColumn('tenantId')
+    })
+    .table('incomingPayments', function (table) {
+      table.dropForeign(['tenantId'])
+      table.dropColumn('tenantId')
+    })
+    .table('outgoingPayments', function (table) {
+      table.dropForeign(['tenantId'])
+      table.dropColumn('tenantId')
+    })
+    .table('walletAddresses', function (table) {
+      table.dropForeign(['tenantId'])
+      table.dropColumn('tenantId')
+    })
+    .table('grants', function (table) {
+      table.dropForeign(['tenantId'])
+      table.dropColumn('tenantId')
+    })
+}

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

@@ -5,6 +5,7 @@ import { OutgoingPayment, OutgoingPaymentState } from './model'
 import { LifecycleError, PaymentError } from './errors'
 import * as lifecycle from './lifecycle'
 import { PaymentMethodHandlerError } from '../../../payment-method/handler/errors'
+import { trace, Span } from '@opentelemetry/api'
 
 // First retry waits 10 seconds, second retry waits 20 (more) seconds, etc.
 export const RETRY_BACKOFF_SECONDS = 10
@@ -15,23 +16,33 @@ const MAX_STATE_ATTEMPTS = 5
 export async function processPendingPayment(
   deps_: ServiceDependencies
 ): Promise<string | undefined> {
-  return deps_.knex.transaction(async (trx) => {
-    const payment = await getPendingPayment(trx)
-    if (!payment) return
-
-    await handlePaymentLifecycle(
-      {
-        ...deps_,
-        knex: trx,
-        logger: deps_.logger.child({
-          payment: payment.id,
-          from_state: payment.state
-        })
-      },
-      payment
-    )
-    return payment.id
-  })
+  const tracer = trace.getTracer('outgoing_payment_worker')
+
+  return tracer.startActiveSpan(
+    'outgoingPaymentLifecycle',
+    async (span: Span) => {
+      const paymentId = await deps_.knex.transaction(async (trx) => {
+        const payment = await getPendingPayment(trx)
+        if (!payment) return
+
+        await handlePaymentLifecycle(
+          {
+            ...deps_,
+            knex: trx,
+            logger: deps_.logger.child({
+              payment: payment.id,
+              from_state: payment.state
+            })
+          },
+          payment
+        )
+        return payment.id
+      })
+
+      span.end()
+      return paymentId
+    }
+  )
 }
 
 // Fetch (and lock) a payment for work.

packages/backend/src/rates/service.test.ts

@@ -6,6 +6,7 @@ import { initIocContainer } from '../'
 import { AppServices } from '../app'
 import { CacheDataStore } from '../middleware/cache/data-stores'
 import { mockRatesApi } from '../tests/rates'
+import { AxiosInstance } from 'axios'
 
 const nock = (global as unknown as { nock: typeof import('nock') }).nock
 
@@ -214,23 +215,27 @@ describe('Rates service', function () {
       expect(apiRequestCount).toBe(2)
     })
 
-    it('prefetches when the cached request is old', async () => {
+    it('returns new rates after cache expires', async () => {
       await expect(service.rates('USD')).resolves.toEqual(usdRates)
-      jest.advanceTimersByTime(exchangeRatesLifetime * 0.5 + 1)
-      // ... cache isn't expired yet, but it will be soon
-      await expect(service.rates('USD')).resolves.toEqual(usdRates)
-      expect(apiRequestCount).toBe(1)
-
-      // Invalidate the cache.
-      jest.advanceTimersByTime(exchangeRatesLifetime * 0.5 + 1)
+      jest.advanceTimersByTime(exchangeRatesLifetime + 1)
       await expect(service.rates('USD')).resolves.toEqual(usdRates)
-      // The prefetch response is promoted to the cache.
       expect(apiRequestCount).toBe(2)
     })
 
-    it('cannot use an expired cache', async () => {
-      await expect(service.rates('USD')).resolves.toEqual(usdRates)
-      jest.advanceTimersByTime(exchangeRatesLifetime + 1)
+    it('returns rates on second request (first one was error)', async () => {
+      jest
+        .spyOn(
+          (service as RatesService & { axios: AxiosInstance }).axios,
+          'get'
+        )
+        .mockImplementationOnce(() => {
+          apiRequestCount++
+          throw new Error()
+        })
+
+      await expect(service.rates('USD')).rejects.toThrow(
+        'Could not fetch rates'
+      )
       await expect(service.rates('USD')).resolves.toEqual(usdRates)
       expect(apiRequestCount).toBe(2)
     })

packages/backend/src/rates/service.ts

@@ -1,5 +1,5 @@
 import { BaseService } from '../shared/baseService'
-import Axios, { AxiosInstance } from 'axios'
+import Axios, { AxiosInstance, isAxiosError } from 'axios'
 import { convert, ConvertOptions } from './util'
 import { createInMemoryDataStore } from '../middleware/cache/data-stores/in-memory'
 import { CacheDataStore } from '../middleware/cache/data-stores'
@@ -74,29 +74,13 @@ class RatesServiceImpl implements RatesService {
   }
 
   private async getRates(baseAssetCode: string): Promise<Rates> {
-    const [cachedRate, cachedExpiry] = await Promise.all([
-      this.cachedRates.get(baseAssetCode),
-      this.cachedRates.getKeyExpiry(baseAssetCode)
-    ])
-
-    if (cachedRate && cachedExpiry) {
-      const isCloseToExpiry =
-        cachedExpiry.getTime() <
-        Date.now() + 0.5 * this.deps.exchangeRatesLifetime
-
-      if (isCloseToExpiry) {
-        this.fetchNewRatesAndCache(baseAssetCode) // don't await, just get new rates for later
-      }
+    const cachedRate = await this.cachedRates.get(baseAssetCode)
 
+    if (cachedRate) {
       return JSON.parse(cachedRate)
     }
 
-    try {
-      return await this.fetchNewRatesAndCache(baseAssetCode)
-    } catch (err) {
-      this.cachedRates.delete(baseAssetCode)
-      throw err
-    }
+    return await this.fetchNewRatesAndCache(baseAssetCode)
   }
 
   private async fetchNewRatesAndCache(baseAssetCode: string): Promise<Rates> {
@@ -106,12 +90,32 @@ class RatesServiceImpl implements RatesService {
       this.inProgressRequests[baseAssetCode] = this.fetchNewRates(baseAssetCode)
     }
 
-    const rates = await this.inProgressRequests[baseAssetCode]
+    try {
+      const rates = await this.inProgressRequests[baseAssetCode]
 
-    delete this.inProgressRequests[baseAssetCode]
+      await this.cachedRates.set(baseAssetCode, JSON.stringify(rates))
+      return rates
+    } catch (err) {
+      const errorMessage = 'Could not fetch rates'
+
+      this.deps.logger.error(
+        {
+          ...(isAxiosError(err)
+            ? {
+                errorMessage: err.message,
+                errorCode: err.code,
+                errorStatus: err.status
+              }
+            : { err }),
+          url: this.deps.exchangeRatesUrl
+        },
+        errorMessage
+      )
 
-    await this.cachedRates.set(baseAssetCode, JSON.stringify(rates))
-    return rates
+      throw new Error(errorMessage)
+    } finally {
+      delete this.inProgressRequests[baseAssetCode]
+    }
   }
 
   private async fetchNewRates(baseAssetCode: string): Promise<Rates> {
@@ -120,12 +124,9 @@ class RatesServiceImpl implements RatesService {
       return { base: baseAssetCode, rates: {} }
     }
 
-    const res = await this.axios
-      .get<Rates>(url, { params: { base: baseAssetCode } })
-      .catch((err) => {
-        this.deps.logger.warn({ err: err.message }, 'price request error')
-        throw err
-      })
+    const res = await this.axios.get<Rates>(url, {
+      params: { base: baseAssetCode }
+    })
 
     const { base, rates } = res.data
     this.checkBaseAsset(base)