Merge pull request #1066 from PayButton/feat/clean-client-payment-instances-after-a-while

feat: job to clean up pending old payments

Author: chedieck

constants/index.ts

@@ -291,3 +291,5 @@ export const TRIGGER_LOG_BATCH_SIZE = 200
 export const PRICES_CONNECTION_BATCH_SIZE = 1_000
 // interactive $transaction timeout in ms (for the single delete + several createMany of prices)
 export const PRICES_CONNECTION_TIMEOUT = 30_000
+
+export const CLIENT_PAYMENT_EXPIRATION_TIME = (7) * (24 * 60 * 60 * 1000) // (number of days) * (24 * 60 * 60 * 1000)

jobs/initJobs.ts

@@ -1,8 +1,8 @@
-import { CURRENT_PRICE_REPEAT_DELAY } from 'constants/index'
+import { CLIENT_PAYMENT_EXPIRATION_TIME, CURRENT_PRICE_REPEAT_DELAY } from 'constants/index'
 import { Queue } from 'bullmq'
 import { redisBullMQ } from 'redis/clientInstance'
 import EventEmitter from 'events'
-import { syncCurrentPricesWorker, syncBlockchainAndPricesWorker } from './workers'
+import { syncCurrentPricesWorker, syncBlockchainAndPricesWorker, cleanupClientPaymentsWorker } from './workers'
 
 EventEmitter.defaultMaxListeners = 20
 
@@ -24,6 +24,22 @@ const main = async (): Promise<void> => {
   const blockchainQueue = new Queue('blockchainSync', { connection: redisBullMQ })
   await blockchainQueue.add('syncBlockchainAndPrices', {}, { jobId: 'syncBlockchainAndPrices' })
   await syncBlockchainAndPricesWorker(blockchainQueue.name)
+
+  const cleanupQueue = new Queue('clientPaymentCleanup', { connection: redisBullMQ })
+
+  await cleanupQueue.add(
+    'cleanupClientPayments',
+    {},
+    {
+      jobId: 'cleanupClientPayments',
+      removeOnFail: false,
+      repeat: {
+        every: CLIENT_PAYMENT_EXPIRATION_TIME
+      }
+    }
+  )
+
+  await cleanupClientPaymentsWorker(cleanupQueue.name)
 }
 
 void main()

jobs/workers.ts

@@ -3,6 +3,7 @@ import { redisBullMQ } from 'redis/clientInstance'
 import { DEFAULT_WORKER_LOCK_DURATION } from 'constants/index'
 import { multiBlockchainClient } from 'services/chronikService'
 import { connectAllTransactionsToPrices } from 'services/transactionService'
+import { cleanupExpiredClientPayments } from 'services/clientPaymentService'
 
 import * as priceService from 'services/priceService'
 
@@ -60,3 +61,26 @@ export const syncBlockchainAndPricesWorker = async (queueName: string): Promise<
     }
   })
 }
+
+export const cleanupClientPaymentsWorker = async (queueName: string): Promise<void> => {
+  const worker = new Worker(
+    queueName,
+    async (job) => {
+      console.log(`[CLIENT_PAYMENT CLEANUP] job ${job.id as string}: running expired payment cleanup...`)
+      await cleanupExpiredClientPayments()
+      console.log('[CLIENT_PAYMENT CLEANUP] cleanup finished.')
+    },
+    {
+      connection: redisBullMQ,
+      lockDuration: DEFAULT_WORKER_LOCK_DURATION
+    }
+  )
+
+  worker.on('completed', job => {
+    console.log(`[CLIENT_PAYMENT CLEANUP] job ${job.id as string}: completed successfully`)
+  })
+
+  worker.on('failed', (job, err) => {
+    console.error(`[CLIENT_PAYMENT CLEANUP] job ${job?.id as string}: FAILED — ${err.message}`)
+  })
+}

pages/api/payments/paymentId/index.ts

@@ -1,5 +1,5 @@
 import { Decimal } from '@prisma/client/runtime/library'
-import { generatePaymentId } from 'services/transactionService'
+import { generatePaymentId } from 'services/clientPaymentService'
 import { parseAddress, parseCreatePaymentIdPOSTRequest } from 'utils/validators'
 import { RESPONSE_MESSAGES } from 'constants/index'
 import { runMiddleware } from 'utils/index'

services/chronikService.ts

@@ -10,10 +10,12 @@ import {
   fetchUnconfirmedTransactions,
   upsertTransaction,
   getSimplifiedTransactions,
-  getSimplifiedTrasaction,
+  getSimplifiedTrasaction
+} from './transactionService'
+import {
   updateClientPaymentStatus,
   getClientPayment
-} from './transactionService'
+} from './clientPaymentService'
 import { Address, Prisma, ClientPaymentStatus } from '@prisma/client'
 import xecaddr from 'xecaddrjs'
 import { getAddressPrefix, satoshisToUnit } from 'utils/index'

services/clientPaymentService.ts

@@ -0,0 +1,82 @@
+import prisma from 'prisma-local/clientInstance'
+import { v4 as uuidv4 } from 'uuid'
+import { multiBlockchainClient } from 'services/chronikService'
+import { Prisma, ClientPaymentStatus } from '@prisma/client'
+import { NETWORK_IDS_FROM_SLUGS, CLIENT_PAYMENT_EXPIRATION_TIME } from 'constants/index'
+import { parseAddress } from 'utils/validators'
+import { addressExists } from './addressService'
+import moment from 'moment'
+
+export const generatePaymentId = async (address: string, amount?: Prisma.Decimal): Promise<string> => {
+  const rawUUID = uuidv4()
+  const cleanUUID = rawUUID.replace(/-/g, '')
+  const status = 'PENDING' as ClientPaymentStatus
+  address = parseAddress(address)
+  const prefix = address.split(':')[0].toLowerCase()
+  const networkId = NETWORK_IDS_FROM_SLUGS[prefix]
+  const isAddressRegistered = await addressExists(address)
+
+  const clientPayment = await prisma.clientPayment.create({
+    data: {
+      address: {
+        connectOrCreate: {
+          where: { address },
+          create: {
+            address,
+            networkId
+          }
+        }
+      },
+      paymentId: cleanUUID,
+      status,
+      amount
+    },
+    include: {
+      address: true
+    }
+  })
+
+  if (!isAddressRegistered) {
+    void multiBlockchainClient.syncAndSubscribeAddresses([clientPayment.address])
+  }
+
+  return clientPayment.paymentId
+}
+
+export const updateClientPaymentStatus = async (paymentId: string, status: ClientPaymentStatus): Promise<void> => {
+  await prisma.clientPayment.update({
+    where: { paymentId },
+    data: { status }
+  })
+}
+
+export const getClientPayment = async (paymentId: string): Promise<Prisma.ClientPaymentGetPayload<{ include: { address: true } }> | null> => {
+  return await prisma.clientPayment.findUnique({
+    where: { paymentId },
+    include: { address: true }
+  })
+}
+
+export const cleanupExpiredClientPayments = async (): Promise<void> => {
+  const cutoff = moment.utc().subtract(CLIENT_PAYMENT_EXPIRATION_TIME, 'milliseconds').toDate()
+
+  const oldPaymentsUnpaid = await prisma.clientPayment.findMany({
+    where: {
+      status: 'PENDING',
+      createdAt: { lt: cutoff }
+    },
+    select: { paymentId: true }
+  })
+
+  if (oldPaymentsUnpaid.length === 0) {
+    console.log('[CLIENT_PAYMENT CLEANUP] no expired pending payments found.')
+    return
+  }
+
+  console.log(`[CLIENT_PAYMENT CLEANUP] deleting ${oldPaymentsUnpaid.length} expired pending payments...`)
+  await prisma.clientPayment.deleteMany({
+    where: {
+      paymentId: { in: oldPaymentsUnpaid.map(p => p.paymentId) }
+    }
+  })
+}

services/transactionService.ts

@@ -1,6 +1,6 @@
 import prisma from 'prisma-local/clientInstance'
-import { Prisma, Transaction, ClientPaymentStatus } from '@prisma/client'
-import { RESPONSE_MESSAGES, USD_QUOTE_ID, CAD_QUOTE_ID, N_OF_QUOTES, UPSERT_TRANSACTION_PRICES_ON_DB_TIMEOUT, SupportedQuotesType, NETWORK_IDS, NETWORK_IDS_FROM_SLUGS, PRICES_CONNECTION_BATCH_SIZE, PRICES_CONNECTION_TIMEOUT } from 'constants/index'
+import { Prisma, Transaction } from '@prisma/client'
+import { RESPONSE_MESSAGES, USD_QUOTE_ID, CAD_QUOTE_ID, N_OF_QUOTES, UPSERT_TRANSACTION_PRICES_ON_DB_TIMEOUT, SupportedQuotesType, NETWORK_IDS, PRICES_CONNECTION_BATCH_SIZE, PRICES_CONNECTION_TIMEOUT } from 'constants/index'
 import { fetchAddressBySubstring, fetchAddressById, fetchAddressesByPaybuttonId, addressExists } from 'services/addressService'
 import { AllPrices, QuoteValues, fetchPricesForNetworkAndTimestamp, flattenTimestamp } from 'services/priceService'
 import _ from 'lodash'
@@ -9,8 +9,6 @@ import { SimplifiedTransaction } from 'ws-service/types'
 import { OpReturnData, parseAddress } from 'utils/validators'
 import { generatePaymentFromTxWithInvoices } from 'redis/paymentCache'
 import { ButtonDisplayData, Payment } from 'redis/types'
-import { v4 as uuidv4 } from 'uuid'
-import { multiBlockchainClient } from 'services/chronikService'
 
 export function getTransactionValue (transaction: TransactionWithPrices | TransactionsWithPaybuttonsAndPrices | SimplifiedTransaction): QuoteValues {
   const ret: QuoteValues = {
@@ -976,53 +974,3 @@ export const fetchDistinctPaymentYearsByUser = async (userId: string): Promise<n
 
   return years.map(y => y.year)
 }
-
-export const generatePaymentId = async (address: string, amount?: Prisma.Decimal): Promise<string> => {
-  const rawUUID = uuidv4()
-  const cleanUUID = rawUUID.replace(/-/g, '')
-  const status = 'PENDING' as ClientPaymentStatus
-  address = parseAddress(address)
-  const prefix = address.split(':')[0].toLowerCase()
-  const networkId = NETWORK_IDS_FROM_SLUGS[prefix]
-  const isAddressRegistered = await addressExists(address)
-
-  const clientPayment = await prisma.clientPayment.create({
-    data: {
-      address: {
-        connectOrCreate: {
-          where: { address },
-          create: {
-            address,
-            networkId
-          }
-        }
-      },
-      paymentId: cleanUUID,
-      status,
-      amount
-    },
-    include: {
-      address: true
-    }
-  })
-
-  if (!isAddressRegistered) {
-    void multiBlockchainClient.syncAndSubscribeAddresses([clientPayment.address])
-  }
-
-  return clientPayment.paymentId
-}
-
-export const updateClientPaymentStatus = async (paymentId: string, status: ClientPaymentStatus): Promise<void> => {
-  await prisma.clientPayment.update({
-    where: { paymentId },
-    data: { status }
-  })
-}
-
-export const getClientPayment = async (paymentId: string): Promise<Prisma.ClientPaymentGetPayload<{ include: { address: true } }> | null> => {
-  return await prisma.clientPayment.findUnique({
-    where: { paymentId },
-    include: { address: true }
-  })
-}

tests/unittests/handleUpdateClientPaymentStatus.test.ts

@@ -4,8 +4,8 @@ import { ChronikBlockchainClient } from '../../services/chronikService'
 
 process.env.WS_AUTH_KEY = 'test-auth-key'
 
-// Mock the transactionService functions
-jest.mock('../../services/transactionService', () => ({
+// Mock the clientPayment functions
+jest.mock('../../services/clientPaymentService', () => ({
   getClientPayment: jest.fn(),
   updateClientPaymentStatus: jest.fn()
 }))
@@ -71,9 +71,9 @@ describe('handleUpdateClientPaymentStatus tests', () => {
 
     // Get the mocked functions
     // eslint-disable-next-line @typescript-eslint/no-var-requires
-    const transactionService = require('../../services/transactionService')
-    mockGetClientPayment = transactionService.getClientPayment
-    mockUpdateClientPaymentStatus = transactionService.updateClientPaymentStatus
+    const clientPaymentService = require('../../services/clientPaymentService')
+    mockGetClientPayment = clientPaymentService.getClientPayment
+    mockUpdateClientPaymentStatus = clientPaymentService.updateClientPaymentStatus
 
     // Clear all mocks before each test
     jest.clearAllMocks()