diff --git a/jobs/cancel_orders.ts b/jobs/cancel_orders.ts index 907c97b2..1d75baa2 100644 --- a/jobs/cancel_orders.ts +++ b/jobs/cancel_orders.ts @@ -6,6 +6,7 @@ import { getUserI18nContext, holdInvoiceExpirationInSecs } from '../util'; import { logger } from '../logger'; import { CommunityContext } from '../bot/modules/community/communityContext'; import * as OrderEvents from '../bot/modules/events/orders'; +import { PerOrderIdMutex } from '../ln/subscribe_invoice'; const cancelOrders = async (bot: HasTelegram) => { try { @@ -30,7 +31,20 @@ const cancelOrders = async (bot: HasTelegram) => { }); for (const order of waitingPaymentOrders) { if (order.status === 'WAITING_PAYMENT') { - await cancelShowHoldInvoice(bot as CommunityContext, order, true); + await PerOrderIdMutex.instance.runExclusive( + String(order._id), + async () => { + const updatedOrder = await Order.findById(order._id); + // In the case the orderId was modified then we don't cancel the order + if (!updatedOrder || updatedOrder.status !== 'WAITING_PAYMENT') + return; + await cancelShowHoldInvoice( + bot as CommunityContext, + updatedOrder, + true, + ); + }, + ); } else { await cancelAddInvoice(bot as CommunityContext, order, true); } diff --git a/ln/subscribe_invoice.ts b/ln/subscribe_invoice.ts index 13217804..abd5a8db 100644 --- a/ln/subscribe_invoice.ts +++ b/ln/subscribe_invoice.ts @@ -8,6 +8,39 @@ import { getUserI18nContext, getEmojiRate, decimalRound } from '../util'; import { logger } from '../logger'; import { HasTelegram } from '../bot/start'; import { IOrder } from '../models/order'; +import { Mutex } from 'async-mutex'; + +type LockCountedMutex = { + lockCount: number; + mutex: Mutex; +}; + +class PerOrderIdMutex { + mutexes: Map = new Map(); + + async runExclusive(orderId: string, callback: () => Promise) { + let mtx: LockCountedMutex; + if (!this.mutexes.has(orderId)) { + mtx = { lockCount: 1, mutex: new Mutex() }; + this.mutexes.set(orderId, mtx); + } else { + mtx = this.mutexes.get(orderId)!; + mtx.lockCount++; + } + let ret: any; + try { + ret = await mtx.mutex.runExclusive(callback); + } finally { + mtx.lockCount--; + if (mtx.lockCount == 0) { + this.mutexes.delete(orderId); + } + } + return ret; + } + + static instance = new PerOrderIdMutex(); +} const subscribeInvoice = async ( bot: HasTelegram, @@ -20,44 +53,64 @@ const subscribeInvoice = async ( if (invoice.is_held && !resub) { const order = await Order.findOne({ hash: invoice.id }); if (order === null) throw new Error('order was not found'); - logger.info( - `Order ${order._id} Invoice with hash: ${id} is being held!`, + await PerOrderIdMutex.instance.runExclusive( + String(order._id), + async () => { + // We need to get an updated version of the order because there is a chance of the cancelOrders coroutine to modify the state of the order + const updatedOrder = await Order.findById(order._id); + if (updatedOrder === null) + throw new Error('order was not found after locking'); + if (updatedOrder.status !== 'WAITING_PAYMENT') { + logger.error( + `Order ${updatedOrder._id} status is not WAITING_PAYMENT on subscribeToInvoice. Actual status: ${updatedOrder.status}`, + ); + return; + } + logger.info( + `Order ${updatedOrder._id} Invoice with hash: ${id} is being held!`, + ); + const buyerUser = await User.findOne({ + _id: updatedOrder.buyer_id, + }); + if (buyerUser === null) throw new Error('buyerUser was not found'); + const sellerUser = await User.findOne({ + _id: updatedOrder.seller_id, + }); + if (sellerUser === null) + throw new Error('sellerUser was not found'); + updatedOrder.status = 'ACTIVE'; + // This is the i18n context we need to pass to the message + const i18nCtxBuyer = await getUserI18nContext(buyerUser); + const i18nCtxSeller = await getUserI18nContext(sellerUser); + if (updatedOrder.type === 'sell') { + await messages.onGoingTakeSellMessage( + bot, + sellerUser, + buyerUser, + updatedOrder, + i18nCtxBuyer, + i18nCtxSeller, + ); + } else if (updatedOrder.type === 'buy') { + updatedOrder.status = 'WAITING_BUYER_INVOICE'; + // We need the seller rating + const stars = getEmojiRate(sellerUser.total_rating); + const roundedRating = decimalRound(sellerUser.total_rating, -1); + const rate = `${roundedRating} ${stars} (${sellerUser.total_reviews})`; + await messages.onGoingTakeBuyMessage( + bot, + sellerUser, + buyerUser, + updatedOrder, + i18nCtxBuyer, + i18nCtxSeller, + rate, + ); + } + updatedOrder.invoice_held_at = new Date(); + await updatedOrder.save(); + }, ); - const buyerUser = await User.findOne({ _id: order.buyer_id }); - if (buyerUser === null) throw new Error('buyerUser was not found'); - const sellerUser = await User.findOne({ _id: order.seller_id }); - if (sellerUser === null) throw new Error('sellerUser was not found'); - order.status = 'ACTIVE'; - // This is the i18n context we need to pass to the message - const i18nCtxBuyer = await getUserI18nContext(buyerUser); - const i18nCtxSeller = await getUserI18nContext(sellerUser); - if (order.type === 'sell') { - await messages.onGoingTakeSellMessage( - bot, - sellerUser, - buyerUser, - order, - i18nCtxBuyer, - i18nCtxSeller, - ); - } else if (order.type === 'buy') { - order.status = 'WAITING_BUYER_INVOICE'; - // We need the seller rating - const stars = getEmojiRate(sellerUser.total_rating); - const roundedRating = decimalRound(sellerUser.total_rating, -1); - const rate = `${roundedRating} ${stars} (${sellerUser.total_reviews})`; - await messages.onGoingTakeBuyMessage( - bot, - sellerUser, - buyerUser, - order, - i18nCtxBuyer, - i18nCtxSeller, - rate, - ); - } - order.invoice_held_at = new Date(); - order.save(); } if (invoice.is_confirmed) { const order = await Order.findOne({ hash: id }); @@ -147,4 +200,4 @@ const payHoldInvoice = async (bot: HasTelegram, order: IOrder) => { } }; -export { subscribeInvoice, payHoldInvoice }; +export { subscribeInvoice, payHoldInvoice, PerOrderIdMutex }; diff --git a/package-lock.json b/package-lock.json index 5b79fa38..2e0f2974 100644 --- a/package-lock.json +++ b/package-lock.json @@ -11,6 +11,7 @@ "dependencies": { "@grammyjs/i18n": "^0.5.1", "@grammyjs/ratelimiter": "^1.1.5", + "async-mutex": "^0.5.0", "axios": "1.9.0", "canvas": "^3.0.0", "crypto": "^1.0.1", @@ -2465,6 +2466,14 @@ "resolved": "https://registry.npmjs.org/async/-/async-3.2.4.tgz", "integrity": "sha512-iAB+JbDEGXhyIUavoDl9WP/Jj106Kz9DEn1DPgYw5ruDn0e3Wgi3sKFm55sASdGBNOQB8F59d9qQ7deqrHA8wQ==" }, + "node_modules/async-mutex": { + "version": "0.5.0", + "resolved": "https://registry.npmjs.org/async-mutex/-/async-mutex-0.5.0.tgz", + "integrity": "sha512-1A94B18jkJ3DYq284ohPxoXbfTA5HsQ7/Mf4DEhcyLx3Bz27Rh59iScbB6EPiP+B+joue6YCxcMXSbFC1tZKwA==", + "dependencies": { + "tslib": "^2.4.0" + } + }, "node_modules/asyncjs-util": { "version": "1.2.12", "resolved": "https://registry.npmjs.org/asyncjs-util/-/asyncjs-util-1.2.12.tgz", @@ -8042,8 +8051,7 @@ "version": "2.8.1", "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.8.1.tgz", "integrity": "sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==", - "license": "0BSD", - "optional": true + "license": "0BSD" }, "node_modules/tstl": { "version": "2.5.16", diff --git a/package.json b/package.json index 281611ed..144ed620 100644 --- a/package.json +++ b/package.json @@ -18,6 +18,7 @@ "dependencies": { "@grammyjs/i18n": "^0.5.1", "@grammyjs/ratelimiter": "^1.1.5", + "async-mutex": "^0.5.0", "axios": "1.9.0", "canvas": "^3.0.0", "crypto": "^1.0.1",