Late payment flow

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);
       }

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<string, LockCountedMutex> = new Map();
+
+  async runExclusive(orderId: string, callback: () => Promise<any>) {
+    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 };

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",