From 4f25325b01d68978bb42374fd9c26790888aa4db Mon Sep 17 00:00:00 2001 From: Lucas Jeffrey Date: Mon, 15 Sep 2025 18:47:06 -0300 Subject: [PATCH 1/8] Preventing the cancelOrders job and the code processing of a hold invoice which just got paid (subscribeToInvoice) to run at the same time --- jobs/cancel_orders.ts | 193 ++++++++++++++++++++-------------------- ln/subscribe_invoice.ts | 113 ++++++++++++----------- package-lock.json | 16 +++- package.json | 1 + 4 files changed, 172 insertions(+), 151 deletions(-) diff --git a/jobs/cancel_orders.ts b/jobs/cancel_orders.ts index 907c97b2..62492ccd 100644 --- a/jobs/cancel_orders.ts +++ b/jobs/cancel_orders.ts @@ -6,107 +6,110 @@ 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 { subscribeToInvoiceMutex } from '../ln/subscribe_invoice'; const cancelOrders = async (bot: HasTelegram) => { - try { - const holdInvoiceTime = new Date(); - holdInvoiceTime.setSeconds( - holdInvoiceTime.getSeconds() - - Number(process.env.HOLD_INVOICE_EXPIRATION_WINDOW), - ); - // We get the orders where the seller didn't pay the hold invoice before expired - // or where the buyer didn't add the invoice - const waitingPaymentOrders = await Order.find({ - $or: [{ status: 'WAITING_PAYMENT' }, { status: 'WAITING_BUYER_INVOICE' }], - $and: [ - { - taken_at: { $lte: holdInvoiceTime }, - $or: [ - { invoice_held_at: { $eq: null } }, - { invoice_held_at: { $lte: holdInvoiceTime } }, - ], - }, - ], - }); - for (const order of waitingPaymentOrders) { - if (order.status === 'WAITING_PAYMENT') { - await cancelShowHoldInvoice(bot as CommunityContext, order, true); - } else { - await cancelAddInvoice(bot as CommunityContext, order, true); + await subscribeToInvoiceMutex.runExclusive(async () => { + try { + const holdInvoiceTime = new Date(); + holdInvoiceTime.setSeconds( + holdInvoiceTime.getSeconds() - + Number(process.env.HOLD_INVOICE_EXPIRATION_WINDOW), + ); + // We get the orders where the seller didn't pay the hold invoice before expired + // or where the buyer didn't add the invoice + const waitingPaymentOrders = await Order.find({ + $or: [{ status: 'WAITING_PAYMENT' }, { status: 'WAITING_BUYER_INVOICE' }], + $and: [ + { + taken_at: { $lte: holdInvoiceTime }, + $or: [ + { invoice_held_at: { $eq: null } }, + { invoice_held_at: { $lte: holdInvoiceTime } }, + ], + }, + ], + }); + for (const order of waitingPaymentOrders) { + if (order.status === 'WAITING_PAYMENT') { + await cancelShowHoldInvoice(bot as CommunityContext, order, true); + } else { + await cancelAddInvoice(bot as CommunityContext, order, true); + } } - } - // We get the expired order where the seller sent the sats but never released the order - // In this case we use another time field, `invoice_held_at` is the time when the - // seller sent the money to the hold invoice, this is an important moment cause - // we don't want to have a CLTV timeout - let orderTime = new Date(); - const holdInvoiceExpiration = holdInvoiceExpirationInSecs(); - orderTime.setSeconds( - orderTime.getSeconds() - holdInvoiceExpiration.expirationTimeInSecs, - ); - const activeOrders = await Order.find({ - invoice_held_at: { $lte: orderTime }, - $or: [ - { - status: 'FIAT_SENT', - }, - ], - admin_warned: false, - }); - for (const order of activeOrders) { - const buyerUser = await User.findOne({ _id: order.buyer_id }); - const sellerUser = await User.findOne({ _id: order.seller_id }); - if (buyerUser === null || sellerUser === null) return; - const i18nCtxBuyer = await getUserI18nContext(buyerUser); - const i18nCtxSeller = await getUserI18nContext(sellerUser); - // Instead of cancel this order we should send this to the admins - // and they decide what to do - await messages.expiredOrderMessage( - bot, - order, - buyerUser, - sellerUser, - i18nCtxBuyer, + // We get the expired order where the seller sent the sats but never released the order + // In this case we use another time field, `invoice_held_at` is the time when the + // seller sent the money to the hold invoice, this is an important moment cause + // we don't want to have a CLTV timeout + let orderTime = new Date(); + const holdInvoiceExpiration = holdInvoiceExpirationInSecs(); + orderTime.setSeconds( + orderTime.getSeconds() - holdInvoiceExpiration.expirationTimeInSecs, ); - // We send messages about the expired order to each party - await messages.toBuyerExpiredOrderMessage(bot, buyerUser, i18nCtxBuyer); - await messages.toSellerExpiredOrderMessage( - bot, - sellerUser, - i18nCtxSeller, + const activeOrders = await Order.find({ + invoice_held_at: { $lte: orderTime }, + $or: [ + { + status: 'FIAT_SENT', + }, + ], + admin_warned: false, + }); + for (const order of activeOrders) { + const buyerUser = await User.findOne({ _id: order.buyer_id }); + const sellerUser = await User.findOne({ _id: order.seller_id }); + if (buyerUser === null || sellerUser === null) return; + const i18nCtxBuyer = await getUserI18nContext(buyerUser); + const i18nCtxSeller = await getUserI18nContext(sellerUser); + // Instead of cancel this order we should send this to the admins + // and they decide what to do + await messages.expiredOrderMessage( + bot, + order, + buyerUser, + sellerUser, + i18nCtxBuyer, + ); + // We send messages about the expired order to each party + await messages.toBuyerExpiredOrderMessage(bot, buyerUser, i18nCtxBuyer); + await messages.toSellerExpiredOrderMessage( + bot, + sellerUser, + i18nCtxSeller, + ); + order.admin_warned = true; + await order.save(); + } + // ============================== + // Now we cancel orders expired + // ============================== + orderTime = new Date(); + let orderExpirationTime = Number( + process.env.ORDER_PUBLISHED_EXPIRATION_WINDOW, ); - order.admin_warned = true; - await order.save(); - } - // ============================== - // Now we cancel orders expired - // ============================== - orderTime = new Date(); - let orderExpirationTime = Number( - process.env.ORDER_PUBLISHED_EXPIRATION_WINDOW, - ); - orderExpirationTime = orderExpirationTime + orderExpirationTime * 0.2; - orderTime.setSeconds(orderTime.getSeconds() - orderExpirationTime); - const expiredOrders = await Order.find({ - invoice_held_at: { $lte: orderTime }, - $or: [ - { - status: 'ACTIVE', - }, - { - status: 'FIAT_SENT', - }, - ], - }); - for (const order of expiredOrders) { - order.status = 'EXPIRED'; - await order.save(); - OrderEvents.orderUpdated(order); - logger.info(`Order Id ${order.id} expired!`); + orderExpirationTime = orderExpirationTime + orderExpirationTime * 0.2; + orderTime.setSeconds(orderTime.getSeconds() - orderExpirationTime); + const expiredOrders = await Order.find({ + invoice_held_at: { $lte: orderTime }, + $or: [ + { + status: 'ACTIVE', + }, + { + status: 'FIAT_SENT', + }, + ], + }); + for (const order of expiredOrders) { + order.status = 'EXPIRED'; + await order.save(); + OrderEvents.orderUpdated(order); + logger.info(`Order Id ${order.id} expired!`); + } + } catch (error) { + logger.error(error); } - } catch (error) { - logger.error(error); - } + }); }; export default cancelOrders; diff --git a/ln/subscribe_invoice.ts b/ln/subscribe_invoice.ts index 13217804..23fffc8d 100644 --- a/ln/subscribe_invoice.ts +++ b/ln/subscribe_invoice.ts @@ -8,6 +8,9 @@ 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'; + +const subscribeToInvoiceMutex = new Mutex; const subscribeInvoice = async ( bot: HasTelegram, @@ -17,62 +20,68 @@ const subscribeInvoice = async ( try { const sub = subscribeToInvoice({ id, lnd }); sub.on('invoice_updated', async invoice => { - 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!`, - ); - 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, + await subscribeToInvoiceMutex.runExclusive(async () => { + if (invoice.is_held && !resub) { + const order = await Order.findOne({ hash: invoice.id }); + if (order === null) throw new Error('order was not found'); + if (order.status !== 'WAITING_PAYMENT') { + logger.error(`Order ${order._id} status is not WAITING_PAYMENT on subscribeToInvoice. Actual status: ${order.status}`); + throw new Error('Order status is not WAITING_PAYMENT'); + } + logger.info( + `Order ${order._id} Invoice with hash: ${id} is being held!`, ); + 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(); + await order.save(); } - order.invoice_held_at = new Date(); - order.save(); - } - if (invoice.is_confirmed) { - const order = await Order.findOne({ hash: id }); - if (order === null) throw new Error('order was not found'); - logger.info( - `Order ${order._id} - Invoice with hash: ${id} was settled!`, - ); - if (order.status === 'FROZEN' && order.is_frozen) { + if (invoice.is_confirmed) { + const order = await Order.findOne({ hash: id }); + if (order === null) throw new Error('order was not found'); logger.info( - `Order ${order._id} - Order was frozen by ${order.action_by}!`, + `Order ${order._id} - Invoice with hash: ${id} was settled!`, ); - return; + if (order.status === 'FROZEN' && order.is_frozen) { + logger.info( + `Order ${order._id} - Order was frozen by ${order.action_by}!`, + ); + return; + } + await payHoldInvoice(bot, order); } - await payHoldInvoice(bot, order); - } + }); }); } catch (error) { logger.error('subscribeInvoice catch: ', error); @@ -147,4 +156,4 @@ const payHoldInvoice = async (bot: HasTelegram, order: IOrder) => { } }; -export { subscribeInvoice, payHoldInvoice }; +export { subscribeInvoice, payHoldInvoice, subscribeToInvoiceMutex }; diff --git a/package-lock.json b/package-lock.json index a956f405..2e0f2974 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,16 +1,17 @@ { "name": "lnp2pbot", - "version": "0.13.3", + "version": "0.14.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "lnp2pbot", - "version": "0.13.3", + "version": "0.14.1", "license": "MIT", "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", From c32f7b9595dc44757f684117cbf916789e3e44a3 Mon Sep 17 00:00:00 2001 From: Lucas Jeffrey Date: Sun, 21 Sep 2025 12:29:47 -0300 Subject: [PATCH 2/8] Improving performance by locking only by order id instead of blocking all invoices for being processed when the invoice is being hold --- jobs/cancel_orders.ts | 200 ++++++++++++++++++++-------------------- ln/subscribe_invoice.ts | 100 +++++++++++++------- 2 files changed, 170 insertions(+), 130 deletions(-) diff --git a/jobs/cancel_orders.ts b/jobs/cancel_orders.ts index 62492ccd..7f838362 100644 --- a/jobs/cancel_orders.ts +++ b/jobs/cancel_orders.ts @@ -6,110 +6,114 @@ 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 { subscribeToInvoiceMutex } from '../ln/subscribe_invoice'; +import { PerOrderIdMutex } from '../ln/subscribe_invoice'; const cancelOrders = async (bot: HasTelegram) => { - await subscribeToInvoiceMutex.runExclusive(async () => { - try { - const holdInvoiceTime = new Date(); - holdInvoiceTime.setSeconds( - holdInvoiceTime.getSeconds() - - Number(process.env.HOLD_INVOICE_EXPIRATION_WINDOW), - ); - // We get the orders where the seller didn't pay the hold invoice before expired - // or where the buyer didn't add the invoice - const waitingPaymentOrders = await Order.find({ - $or: [{ status: 'WAITING_PAYMENT' }, { status: 'WAITING_BUYER_INVOICE' }], - $and: [ - { - taken_at: { $lte: holdInvoiceTime }, - $or: [ - { invoice_held_at: { $eq: null } }, - { invoice_held_at: { $lte: holdInvoiceTime } }, - ], - }, - ], - }); - for (const order of waitingPaymentOrders) { - if (order.status === 'WAITING_PAYMENT') { - await cancelShowHoldInvoice(bot as CommunityContext, order, true); - } else { - await cancelAddInvoice(bot as CommunityContext, order, true); - } + try { + logger.info("CancelOrders job"); + const holdInvoiceTime = new Date(); + holdInvoiceTime.setSeconds( + holdInvoiceTime.getSeconds() - + Number(process.env.HOLD_INVOICE_EXPIRATION_WINDOW), + ); + // We get the orders where the seller didn't pay the hold invoice before expired + // or where the buyer didn't add the invoice + const waitingPaymentOrders = await Order.find({ + $or: [{ status: 'WAITING_PAYMENT' }, { status: 'WAITING_BUYER_INVOICE' }], + $and: [ + { + taken_at: { $lte: holdInvoiceTime }, + $or: [ + { invoice_held_at: { $eq: null } }, + { invoice_held_at: { $lte: holdInvoiceTime } }, + ], + }, + ], + }); + for (const order of waitingPaymentOrders) { + if (order.status === 'WAITING_PAYMENT') { + 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); } - // We get the expired order where the seller sent the sats but never released the order - // In this case we use another time field, `invoice_held_at` is the time when the - // seller sent the money to the hold invoice, this is an important moment cause - // we don't want to have a CLTV timeout - let orderTime = new Date(); - const holdInvoiceExpiration = holdInvoiceExpirationInSecs(); - orderTime.setSeconds( - orderTime.getSeconds() - holdInvoiceExpiration.expirationTimeInSecs, + } + // We get the expired order where the seller sent the sats but never released the order + // In this case we use another time field, `invoice_held_at` is the time when the + // seller sent the money to the hold invoice, this is an important moment cause + // we don't want to have a CLTV timeout + let orderTime = new Date(); + const holdInvoiceExpiration = holdInvoiceExpirationInSecs(); + orderTime.setSeconds( + orderTime.getSeconds() - holdInvoiceExpiration.expirationTimeInSecs, + ); + const activeOrders = await Order.find({ + invoice_held_at: { $lte: orderTime }, + $or: [ + { + status: 'FIAT_SENT', + }, + ], + admin_warned: false, + }); + for (const order of activeOrders) { + const buyerUser = await User.findOne({ _id: order.buyer_id }); + const sellerUser = await User.findOne({ _id: order.seller_id }); + if (buyerUser === null || sellerUser === null) return; + const i18nCtxBuyer = await getUserI18nContext(buyerUser); + const i18nCtxSeller = await getUserI18nContext(sellerUser); + // Instead of cancel this order we should send this to the admins + // and they decide what to do + await messages.expiredOrderMessage( + bot, + order, + buyerUser, + sellerUser, + i18nCtxBuyer, ); - const activeOrders = await Order.find({ - invoice_held_at: { $lte: orderTime }, - $or: [ - { - status: 'FIAT_SENT', - }, - ], - admin_warned: false, - }); - for (const order of activeOrders) { - const buyerUser = await User.findOne({ _id: order.buyer_id }); - const sellerUser = await User.findOne({ _id: order.seller_id }); - if (buyerUser === null || sellerUser === null) return; - const i18nCtxBuyer = await getUserI18nContext(buyerUser); - const i18nCtxSeller = await getUserI18nContext(sellerUser); - // Instead of cancel this order we should send this to the admins - // and they decide what to do - await messages.expiredOrderMessage( - bot, - order, - buyerUser, - sellerUser, - i18nCtxBuyer, - ); - // We send messages about the expired order to each party - await messages.toBuyerExpiredOrderMessage(bot, buyerUser, i18nCtxBuyer); - await messages.toSellerExpiredOrderMessage( - bot, - sellerUser, - i18nCtxSeller, - ); - order.admin_warned = true; - await order.save(); - } - // ============================== - // Now we cancel orders expired - // ============================== - orderTime = new Date(); - let orderExpirationTime = Number( - process.env.ORDER_PUBLISHED_EXPIRATION_WINDOW, + // We send messages about the expired order to each party + await messages.toBuyerExpiredOrderMessage(bot, buyerUser, i18nCtxBuyer); + await messages.toSellerExpiredOrderMessage( + bot, + sellerUser, + i18nCtxSeller, ); - orderExpirationTime = orderExpirationTime + orderExpirationTime * 0.2; - orderTime.setSeconds(orderTime.getSeconds() - orderExpirationTime); - const expiredOrders = await Order.find({ - invoice_held_at: { $lte: orderTime }, - $or: [ - { - status: 'ACTIVE', - }, - { - status: 'FIAT_SENT', - }, - ], - }); - for (const order of expiredOrders) { - order.status = 'EXPIRED'; - await order.save(); - OrderEvents.orderUpdated(order); - logger.info(`Order Id ${order.id} expired!`); - } - } catch (error) { - logger.error(error); + order.admin_warned = true; + await order.save(); + } + // ============================== + // Now we cancel orders expired + // ============================== + orderTime = new Date(); + let orderExpirationTime = Number( + process.env.ORDER_PUBLISHED_EXPIRATION_WINDOW, + ); + orderExpirationTime = orderExpirationTime + orderExpirationTime * 0.2; + orderTime.setSeconds(orderTime.getSeconds() - orderExpirationTime); + const expiredOrders = await Order.find({ + invoice_held_at: { $lte: orderTime }, + $or: [ + { + status: 'ACTIVE', + }, + { + status: 'FIAT_SENT', + }, + ], + }); + for (const order of expiredOrders) { + order.status = 'EXPIRED'; + await order.save(); + OrderEvents.orderUpdated(order); + logger.info(`Order Id ${order.id} expired!`); } - }); + } catch (error) { + logger.error(error); + } }; export default cancelOrders; diff --git a/ln/subscribe_invoice.ts b/ln/subscribe_invoice.ts index 23fffc8d..56656e6c 100644 --- a/ln/subscribe_invoice.ts +++ b/ln/subscribe_invoice.ts @@ -7,11 +7,43 @@ import * as ordersActions from '../bot/ordersActions'; import { getUserI18nContext, getEmojiRate, decimalRound } from '../util'; import { logger } from '../logger'; import { HasTelegram } from '../bot/start'; -import { IOrder } from '../models/order'; +import order, { IOrder } from '../models/order'; import {Mutex} from 'async-mutex'; const subscribeToInvoiceMutex = new 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, id: string, @@ -20,36 +52,39 @@ const subscribeInvoice = async ( try { const sub = subscribeToInvoice({ id, lnd }); sub.on('invoice_updated', async invoice => { - await subscribeToInvoiceMutex.runExclusive(async () => { - if (invoice.is_held && !resub) { - const order = await Order.findOne({ hash: invoice.id }); - if (order === null) throw new Error('order was not found'); - if (order.status !== 'WAITING_PAYMENT') { - logger.error(`Order ${order._id} status is not WAITING_PAYMENT on subscribeToInvoice. Actual status: ${order.status}`); + if (invoice.is_held && !resub) { + const order = await Order.findOne({ hash: invoice.id }); + if (order === null) throw new Error('order was not found'); + 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}`); throw new Error('Order status is not WAITING_PAYMENT'); } logger.info( - `Order ${order._id} Invoice with hash: ${id} is being held!`, + `Order ${updatedOrder._id} Invoice with hash: ${id} is being held!`, ); - const buyerUser = await User.findOne({ _id: order.buyer_id }); + 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: order.seller_id }); + const sellerUser = await User.findOne({ _id: updatedOrder.seller_id }); if (sellerUser === null) throw new Error('sellerUser was not found'); - order.status = 'ACTIVE'; + 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 (order.type === 'sell') { + if (updatedOrder.type === 'sell') { await messages.onGoingTakeSellMessage( bot, sellerUser, buyerUser, - order, + updatedOrder, i18nCtxBuyer, i18nCtxSeller, ); - } else if (order.type === 'buy') { - order.status = 'WAITING_BUYER_INVOICE'; + } 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); @@ -58,30 +93,31 @@ const subscribeInvoice = async ( bot, sellerUser, buyerUser, - order, + updatedOrder, i18nCtxBuyer, i18nCtxSeller, rate, ); } - order.invoice_held_at = new Date(); - await order.save(); - } - if (invoice.is_confirmed) { - const order = await Order.findOne({ hash: id }); - if (order === null) throw new Error('order was not found'); + updatedOrder.invoice_held_at = new Date(); + await updatedOrder.save(); + + }); + } + if (invoice.is_confirmed) { + const order = await Order.findOne({ hash: id }); + if (order === null) throw new Error('order was not found'); + logger.info( + `Order ${order._id} - Invoice with hash: ${id} was settled!`, + ); + if (order.status === 'FROZEN' && order.is_frozen) { logger.info( - `Order ${order._id} - Invoice with hash: ${id} was settled!`, + `Order ${order._id} - Order was frozen by ${order.action_by}!`, ); - if (order.status === 'FROZEN' && order.is_frozen) { - logger.info( - `Order ${order._id} - Order was frozen by ${order.action_by}!`, - ); - return; - } - await payHoldInvoice(bot, order); + return; } - }); + await payHoldInvoice(bot, order); + } }); } catch (error) { logger.error('subscribeInvoice catch: ', error); @@ -156,4 +192,4 @@ const payHoldInvoice = async (bot: HasTelegram, order: IOrder) => { } }; -export { subscribeInvoice, payHoldInvoice, subscribeToInvoiceMutex }; +export { subscribeInvoice, payHoldInvoice, PerOrderIdMutex }; From cf24cdf5e2cac688c96dbeb3055b5752bbf7ea86 Mon Sep 17 00:00:00 2001 From: Lucas Jeffrey Date: Sun, 21 Sep 2025 12:39:20 -0300 Subject: [PATCH 3/8] Deleting the subscribeToInvoicesMutex because is no longer being used --- ln/subscribe_invoice.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/ln/subscribe_invoice.ts b/ln/subscribe_invoice.ts index 56656e6c..4eb4570a 100644 --- a/ln/subscribe_invoice.ts +++ b/ln/subscribe_invoice.ts @@ -10,8 +10,6 @@ import { HasTelegram } from '../bot/start'; import order, { IOrder } from '../models/order'; import {Mutex} from 'async-mutex'; -const subscribeToInvoiceMutex = new Mutex; - type LockCountedMutex = { lockCount: number, mutex: Mutex From e10841fbe66f2b50548c15a68c361726bbcf9b6b Mon Sep 17 00:00:00 2001 From: Lucas Jeffrey Date: Sat, 4 Oct 2025 21:10:40 -0300 Subject: [PATCH 4/8] Removing debug logs on cancelOrders job (late-payment-flow) --- jobs/cancel_orders.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/jobs/cancel_orders.ts b/jobs/cancel_orders.ts index 7f838362..2b1a1344 100644 --- a/jobs/cancel_orders.ts +++ b/jobs/cancel_orders.ts @@ -10,7 +10,6 @@ import { PerOrderIdMutex } from '../ln/subscribe_invoice'; const cancelOrders = async (bot: HasTelegram) => { try { - logger.info("CancelOrders job"); const holdInvoiceTime = new Date(); holdInvoiceTime.setSeconds( holdInvoiceTime.getSeconds() - From 82a22636d720a02dd5dfa3a2fe1eab255ce09f95 Mon Sep 17 00:00:00 2001 From: Luquitasjeffrey <105951354+Luquitasjeffrey@users.noreply.github.com> Date: Thu, 9 Oct 2025 17:24:37 -0300 Subject: [PATCH 5/8] Remove import of unused variable "order" Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> --- ln/subscribe_invoice.ts | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/ln/subscribe_invoice.ts b/ln/subscribe_invoice.ts index 4eb4570a..83a8e7bb 100644 --- a/ln/subscribe_invoice.ts +++ b/ln/subscribe_invoice.ts @@ -7,9 +7,8 @@ import * as ordersActions from '../bot/ordersActions'; import { getUserI18nContext, getEmojiRate, decimalRound } from '../util'; import { logger } from '../logger'; import { HasTelegram } from '../bot/start'; -import order, { IOrder } from '../models/order'; -import {Mutex} from 'async-mutex'; - +import { IOrder } from '../models/order'; +import { Mutex } from 'async-mutex'; type LockCountedMutex = { lockCount: number, mutex: Mutex From 37f288c7a42bdb6f38dfd4cc490124e71adb4a33 Mon Sep 17 00:00:00 2001 From: Luquitasjeffrey <105951354+Luquitasjeffrey@users.noreply.github.com> Date: Thu, 9 Oct 2025 17:25:31 -0300 Subject: [PATCH 6/8] Update ln/subscribe_invoice.ts Fix the comment spacing to keep eslint happy. Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> --- ln/subscribe_invoice.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ln/subscribe_invoice.ts b/ln/subscribe_invoice.ts index 83a8e7bb..78ae8ac6 100644 --- a/ln/subscribe_invoice.ts +++ b/ln/subscribe_invoice.ts @@ -53,7 +53,7 @@ const subscribeInvoice = async ( const order = await Order.findOne({ hash: invoice.id }); if (order === null) throw new Error('order was not found'); 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 + // 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') { From 632e8ef5343208b200b21ff73edc21796252dbd7 Mon Sep 17 00:00:00 2001 From: Lucas Jeffrey Date: Thu, 9 Oct 2025 18:34:28 -0300 Subject: [PATCH 7/8] Instead of throwing an error when the status of the order is not WAITING_PAYMENT, just debug the message and return without exceptions --- ln/subscribe_invoice.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ln/subscribe_invoice.ts b/ln/subscribe_invoice.ts index 78ae8ac6..fe286f0c 100644 --- a/ln/subscribe_invoice.ts +++ b/ln/subscribe_invoice.ts @@ -9,6 +9,7 @@ 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 @@ -58,7 +59,7 @@ const subscribeInvoice = async ( 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}`); - throw new Error('Order status is not WAITING_PAYMENT'); + return; } logger.info( `Order ${updatedOrder._id} Invoice with hash: ${id} is being held!`, From 60852d9edd8a721ea3b94bc51758e6306c8263c1 Mon Sep 17 00:00:00 2001 From: Lucas Jeffrey Date: Thu, 9 Oct 2025 18:36:15 -0300 Subject: [PATCH 8/8] Formatting the code after running 'npm run format' --- jobs/cancel_orders.ts | 20 +++++-- ln/subscribe_invoice.ts | 120 ++++++++++++++++++++++------------------ 2 files changed, 79 insertions(+), 61 deletions(-) diff --git a/jobs/cancel_orders.ts b/jobs/cancel_orders.ts index 2b1a1344..1d75baa2 100644 --- a/jobs/cancel_orders.ts +++ b/jobs/cancel_orders.ts @@ -31,12 +31,20 @@ const cancelOrders = async (bot: HasTelegram) => { }); for (const order of waitingPaymentOrders) { if (order.status === 'WAITING_PAYMENT') { - 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); - }); + 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 fe286f0c..abd5a8db 100644 --- a/ln/subscribe_invoice.ts +++ b/ln/subscribe_invoice.ts @@ -11,17 +11,17 @@ import { IOrder } from '../models/order'; import { Mutex } from 'async-mutex'; type LockCountedMutex = { - lockCount: number, - mutex: Mutex -} + lockCount: number; + mutex: Mutex; +}; class PerOrderIdMutex { - mutexes: Map = new Map; + mutexes: Map = new Map(); - async runExclusive(orderId: string, callback: ()=>Promise) { + async runExclusive(orderId: string, callback: () => Promise) { let mtx: LockCountedMutex; if (!this.mutexes.has(orderId)) { - mtx = {lockCount: 1, mutex: new Mutex}; + mtx = { lockCount: 1, mutex: new Mutex() }; this.mutexes.set(orderId, mtx); } else { mtx = this.mutexes.get(orderId)!; @@ -32,14 +32,14 @@ class PerOrderIdMutex { ret = await mtx.mutex.runExclusive(callback); } finally { mtx.lockCount--; - if (mtx.lockCount==0) { + if (mtx.lockCount == 0) { this.mutexes.delete(orderId); } } return ret; } - static instance = new PerOrderIdMutex; + static instance = new PerOrderIdMutex(); } const subscribeInvoice = async ( @@ -53,54 +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'); - 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, + 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!`, ); - } - updatedOrder.invoice_held_at = new Date(); - await updatedOrder.save(); - - }); + 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(); + }, + ); } if (invoice.is_confirmed) { const order = await Order.findOne({ hash: id });