try catch around queue processing

2 files changed, 88 insertions, 74 deletions

diff --git a/routes/bitcoin/receive.js b/routes/bitcoin/receive.js
index 9b01fc6a..77aabd68 100644
--- a/routes/bitcoin/receive.js
+++ b/routes/bitcoin/receive.js
@@ -25,7 +25,7 @@ zmqRawTx.on("message", async (topic, message, sequence) => {
   if (payments.includes(hash)) return;
 
   Promise.all(
-    tx.outs.map(async o => {
+    tx.outs.map(async (o) => {
       const { value } = o;
 
       let address;
@@ -40,16 +40,16 @@ zmqRawTx.on("message", async (topic, message, sequence) => {
 
         let user = await db.User.findOne({
           where: {
-            username: addresses[address]
-          }
+            username: addresses[address],
+          },
         });
 
         const invoice = await db.Invoice.findOne({
           where: {
             user_id: user.id,
-            network: "BTC"
+            network: "BTC",
           },
-          order: [["id", "DESC"]]
+          order: [["id", "DESC"]],
         });
 
         const currency = invoice ? invoice.currency : user.currency;
@@ -61,8 +61,8 @@ zmqRawTx.on("message", async (topic, message, sequence) => {
         const account = await db.Account.findOne({
           where: {
             user_id: user.id,
-            asset: config.liquid.btcasset
-          }
+            asset: config.liquid.btcasset,
+          },
         });
 
         account.pending += value;
@@ -81,10 +81,10 @@ zmqRawTx.on("message", async (topic, message, sequence) => {
         for (let i = 0; i < tx.ins.length; i++) {
           let { hash, index } = tx.ins[i];
           hash = reverse(hash).toString("hex");
-          let hex = await bc.getRawTransaction(hash.toString('hex'));
+          let hex = await bc.getRawTransaction(hash.toString("hex"));
           let inputTx = bitcoin.Transaction.fromHex(hex);
           totalInputs += inputTx.outs[index].value;
-        } 
+        }
         let fee = totalInputs - totalOutputs;
 
         let payment = await db.Payment.create({
@@ -99,7 +99,7 @@ zmqRawTx.on("message", async (topic, message, sequence) => {
           tip,
           confirmed,
           address,
-          network: "BTC"
+          network: "BTC",
         });
         payment = payment.get({ plain: true });
         payment.account = account.get({ plain: true });
@@ -117,46 +117,50 @@ let queue = {};
 
 zmqRawBlock.on("message", async (topic, message, sequence) => {
   const payments = await db.Payment.findAll({
-    where: { confirmed: false }
+    where: { confirmed: false },
   });
 
-  const hashes = payments.map(p => p.hash);
+  const hashes = payments.map((p) => p.hash);
 
   let block = bitcoin.Block.fromHex(message.toString("hex"));
-  block.transactions.map(tx => {
+  block.transactions.map((tx) => {
     let hash = reverse(tx.getHash()).toString("hex");
     if (hashes.includes(hash)) queue[hash] = 1;
   });
 });
 
 setInterval(async () => {
-  let arr = Object.keys(queue);
-  for (let i = 0; i < arr.length; i++) {
-    let hash = arr[i];
+  try {
+    let arr = Object.keys(queue);
+    for (let i = 0; i < arr.length; i++) {
+      let hash = arr[i];
 
-    let p = await db.Payment.findOne({
-      where: { hash, confirmed: 0, received: 1 },
-      include: {
-        model: db.Account,
-        as: "account"
-      }
-    });
+      let p = await db.Payment.findOne({
+        where: { hash, confirmed: 0, received: 1 },
+        include: {
+          model: db.Account,
+          as: "account",
+        },
+      });
 
-    if (p) {
-      p.confirmed = 1;
-      p.account.balance += p.amount + p.tip;
-      p.account.pending -= Math.min(p.account.pending, p.amount + p.tip);
+      if (p) {
+        p.confirmed = 1;
+        p.account.balance += p.amount + p.tip;
+        p.account.pending -= Math.min(p.account.pending, p.amount + p.tip);
 
-      await p.account.save();
-      await p.save();
+        await p.account.save();
+        await p.save();
 
-      let user = await getUserById(p.user_id);
-      emit(user.username, "user", user);
-      emit(user.username, "payment", p);
-      l.info("bitcoin confirmed", user.username, p.amount, p.tip);
-      delete queue[hash];
-    } else {
-      l.warn("Couldn't find payment", hash);
-    } 
+        let user = await getUserById(p.user_id);
+        emit(user.username, "user", user);
+        emit(user.username, "payment", p);
+        l.info("bitcoin confirmed", user.username, p.amount, p.tip);
+        delete queue[hash];
+      } else {
+        l.warn("Couldn't find payment", hash);
+      }
+    }
+  } catch (e) {
+    l.error("problem processing queued bitcoin transaction", e.message);
   }
 }, 1000);
diff --git a/routes/liquid/receive.js b/routes/liquid/receive.js
index 3ee2eadb..a7d02829 100644
--- a/routes/liquid/receive.js
+++ b/routes/liquid/receive.js
@@ -20,7 +20,7 @@ zmqRawTx.on("message", async (topic, message, sequence) => {
   if (payments.includes(blinded.txid)) return;
 
   Promise.all(
-    tx.vout.map(async o => {
+    tx.vout.map(async (o) => {
       if (!(o.scriptPubKey && o.scriptPubKey.addresses)) return;
 
       const { asset } = o;
@@ -30,20 +30,19 @@ zmqRawTx.on("message", async (topic, message, sequence) => {
       if (Object.keys(addresses).includes(address)) {
         let user = await db.User.findOne({
           where: {
-            username: addresses[address]
-          }
+            username: addresses[address],
+          },
         });
 
-
         let confirmed = 0;
 
         let params = {
           user_id: user.id,
-          asset
+          asset,
         };
 
         let account = await db.Account.findOne({
-          where: params
+          where: params,
         });
 
         if (account) {
@@ -54,10 +53,11 @@ zmqRawTx.on("message", async (topic, message, sequence) => {
           let ticker = asset.substr(0, 3).toUpperCase();
           let precision = 8;
 
-          const assets = (await axios.get("https://assets.blockstream.info/")).data;
+          const assets = (await axios.get("https://assets.blockstream.info/"))
+            .data;
 
           if (assets[asset]) {
-            ({ticker, precision, name } = assets[asset]);
+            ({ ticker, precision, name } = assets[asset]);
           }
 
           params = { ...params, ...{ ticker, precision, name } };
@@ -70,9 +70,9 @@ zmqRawTx.on("message", async (topic, message, sequence) => {
           await lq.walletPassphrase(config.liquid.walletpass, 300);
 
         user.confidential = await lq.getNewAddress();
-        user.liquid = (await lq.getAddressInfo(
-          user.confidential
-        )).unconfidential;
+        user.liquid = (
+          await lq.getAddressInfo(user.confidential)
+        ).unconfidential;
 
         await user.save();
 
@@ -83,9 +83,9 @@ zmqRawTx.on("message", async (topic, message, sequence) => {
           invoice = await db.Invoice.findOne({
             where: {
               user_id: user.id,
-              network: 'LBTC',
+              network: "LBTC",
             },
-            order: [["id", "DESC"]]
+            order: [["id", "DESC"]],
           });
         }
 
@@ -104,7 +104,7 @@ zmqRawTx.on("message", async (topic, message, sequence) => {
           tip,
           confirmed,
           address,
-          network: "LBTC"
+          network: "LBTC",
         });
 
         payments.push(blinded.txid);
@@ -124,46 +124,56 @@ let queue = {};
 
 zmqRawBlock.on("message", async (topic, message, sequence) => {
   const payments = await db.Payment.findAll({
-    where: { confirmed: 0 }
+    where: { confirmed: 0 },
   });
 
-  const hashes = payments.map(p => p.hash);
+  const hashes = payments.map((p) => p.hash);
 
   const block = elements.Block.fromHex(message.toString("hex"), true);
   const hash = await lq.getBlockHash(block.height);
   const json = await lq.getBlock(hash, 2);
 
-  json.tx.map(tx => {
+  json.tx.map((tx) => {
     if (hashes.includes(tx.txid)) queue[tx.txid] = 1;
   });
 });
 
 setInterval(async () => {
-  let arr = Object.keys(queue);
+  try {
+    let arr = Object.keys(queue);
 
-  for (let i = 0; i < arr.length; i++) {
-    let hash = arr[i];
+    for (let i = 0; i < arr.length; i++) {
+      let hash = arr[i];
 
-    let p = await db.Payment.findOne({
-      where: { hash, confirmed: 0, received: 1 },
-      include: {
-        model: db.Account,
-        as: "account"
-      }
-    });
+      let p = await db.Payment.findOne({
+        where: { hash, confirmed: 0, received: 1 },
+        include: {
+          model: db.Account,
+          as: "account",
+        },
+      });
 
-    p.confirmed = 1;
-    p.account.balance += p.amount + p.tip;
-    p.account.pending -= Math.min(p.account.pending, p.amount + p.tip);
+      p.confirmed = 1;
+      p.account.balance += p.amount + p.tip;
+      p.account.pending -= Math.min(p.account.pending, p.amount + p.tip);
 
-    await p.account.save();
-    await p.save();
+      await p.account.save();
+      await p.save();
 
-    let user = await getUserById(p.user_id);
-    emit(user.username, "user", user);
-    emit(user.username, "payment", p);
-    l.info("liquid confirmed", user.username, p.account.asset, p.amount, p.tip);
+      let user = await getUserById(p.user_id);
+      emit(user.username, "user", user);
+      emit(user.username, "payment", p);
+      l.info(
+        "liquid confirmed",
+        user.username,
+        p.account.asset,
+        p.amount,
+        p.tip
+      );
 
-    delete queue[hash];
+      delete queue[hash];
+    }
+  } catch (e) {
+    l.error("problem processing queued liquid transaction", e.message);
   }
 }, 1000);