zmq 6 syntax

1 files changed, 112 insertions, 108 deletions

diff --git a/routes/bitcoin/receive.js b/routes/bitcoin/receive.js
index 67ccdb45..572987cc 100644
--- a/routes/bitcoin/receive.js
+++ b/routes/bitcoin/receive.js
@@ -1,15 +1,15 @@
 const reverse = require("buffer-reverse");
-const zmq = require("zeromq/v5-compat");
+const zmq = require("zeromq");
 const { Op } = require("sequelize");
 const { fromBase58 } = require("bip32");
 
 const bitcoin = require("bitcoinjs-lib");
 
-const zmqRawBlock = zmq.socket("sub");
+const zmqRawBlock = new zmq.Subscriber();
 zmqRawBlock.connect(config.bitcoin.zmqrawblock);
-zmqRawBlock.subscribe("rawblock");
+zmqRawBlock.subscribe("rawtx");
 
-const zmqRawTx = zmq.socket("sub");
+const zmqRawTx = new zmq.Subscriber();
 zmqRawTx.connect(config.bitcoin.zmqrawtx);
 zmqRawTx.subscribe("rawtx");
 
@@ -18,124 +18,128 @@ const network =
     config.bitcoin.network === "mainnet" ? "bitcoin" : config.bitcoin.network
   ];
 
-zmqRawTx.on("message", async (topic, message, sequence) => {
-  const hex = message.toString("hex");
-  let tx = bitcoin.Transaction.fromHex(message);
+let queue = {};
 
-  let hash = reverse(tx.getHash()).toString("hex");
+(async () => {
+  for await (const [topic, message] of zmqRawTx) {
+    const hex = message.toString("hex");
+    let tx = bitcoin.Transaction.fromHex(message);
 
-  if (payments.includes(hash)) return;
+    let hash = reverse(tx.getHash()).toString("hex");
 
-  Promise.all(
-    tx.outs.map(async o => {
-      try {
-        const { value } = o;
+    if (payments.includes(hash)) return;
 
-        let address;
+    Promise.all(
+      tx.outs.map(async (o) => {
         try {
-          address = bitcoin.address.fromOutputScript(o.script, network);
-        } catch (e) {
-          return;
-        }
+          const { value } = o;
 
-        if (
-          Object.keys(addresses).includes(address) &&
-          !change.includes(address)
-        ) {
-          payments.push(hash);
+          let address;
+          try {
+            address = bitcoin.address.fromOutputScript(o.script, network);
+          } catch (e) {
+            return;
+          }
 
-          let user = await db.User.findOne({
-            where: {
-              username: addresses[address]
-            }
-          });
+          if (
+            Object.keys(addresses).includes(address) &&
+            !change.includes(address)
+          ) {
+            payments.push(hash);
 
-          const invoice = await db.Invoice.findOne({
-            where: {
-              user_id: user.id,
-              network: "bitcoin"
-            },
-            order: [["id", "DESC"]],
-            include: {
-              model: db.Account,
-              as: "account"
-            }
-          });
+            let user = await db.User.findOne({
+              where: {
+                username: addresses[address],
+              },
+            });
 
-          if (!invoice) return;
+            const invoice = await db.Invoice.findOne({
+              where: {
+                user_id: user.id,
+                network: "bitcoin",
+              },
+              order: [["id", "DESC"]],
+              include: {
+                model: db.Account,
+                as: "account",
+              },
+            });
 
-          const currency = invoice ? invoice.currency : user.currency;
-          const rate = invoice ? invoice.rate : app.get("rates")[user.currency];
-          const tip = invoice ? invoice.tip : 0;
-          const memo = invoice ? invoice.memo : "";
+            if (!invoice) return;
 
-          let confirmed = false;
+            const currency = invoice ? invoice.currency : user.currency;
+            const rate = invoice
+              ? invoice.rate
+              : app.get("rates")[user.currency];
+            const tip = invoice ? invoice.tip : 0;
+            const memo = invoice ? invoice.memo : "";
 
-          let { account } = invoice;
+            let confirmed = false;
 
-          account.pending += value;
-          await account.save();
+            let { account } = invoice;
 
-          if (config.bitcoin.walletpass)
-            await bc.walletPassphrase(config.bitcoin.walletpass, 300);
+            account.pending += value;
+            await account.save();
 
-          let totalOutputs = tx.outs.reduce((a, b) => a + b.value, 0);
-          let totalInputs = 0;
-          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);
-            let inputTx = bitcoin.Transaction.fromHex(hex);
-            totalInputs += inputTx.outs[index].value;
-          }
-          let fee = totalInputs - totalOutputs;
+            if (config.bitcoin.walletpass)
+              await bc.walletPassphrase(config.bitcoin.walletpass, 300);
 
-          let payment = await db.Payment.create({
-            account_id: account.id,
-            user_id: user.id,
-            hash,
-            fee,
-            memo,
-            amount: value - tip,
-            currency,
-            rate,
-            received: true,
-            tip,
-            confirmed,
-            address,
-            network: "bitcoin",
-            invoice_id: invoice.id
-          });
-          payment = payment.get({ plain: true });
-          payment.account = account.get({ plain: true });
+            let totalOutputs = tx.outs.reduce((a, b) => a + b.value, 0);
+            let totalInputs = 0;
+            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);
+              let inputTx = bitcoin.Transaction.fromHex(hex);
+              totalInputs += inputTx.outs[index].value;
+            }
+            let fee = totalInputs - totalOutputs;
 
-          emit(user.username, "account", account);
+            let payment = await db.Payment.create({
+              account_id: account.id,
+              user_id: user.id,
+              hash,
+              fee,
+              memo,
+              amount: value - tip,
+              currency,
+              rate,
+              received: true,
+              tip,
+              confirmed,
+              address,
+              network: "bitcoin",
+              invoice_id: invoice.id,
+            });
+            payment = payment.get({ plain: true });
+            payment.account = account.get({ plain: true });
 
-          emit(user.username, "payment", payment);
-          l.info("bitcoin detected", user.username, value);
-          notify(user, `${value} SAT payment detected`);
-          callWebhook(invoice, payment);
-        }
-      } catch (e) {
-        console.log(e);
-      }
-    })
-  );
-});
+            emit(user.username, "account", account);
 
-let queue = {};
+            emit(user.username, "payment", payment);
+            l.info("bitcoin detected", user.username, value);
+            notify(user, `${value} SAT payment detected`);
+            callWebhook(invoice, payment);
+          }
+        } catch (e) {
+          console.log(e);
+        }
+      })
+    );
+  }
 
-zmqRawBlock.on("message", async (topic, message, sequence) => {
-  const payments = await db.Payment.findAll({
-    where: { confirmed: false }
-  });
+  for await (const [topic, message] of zmqRawBlock) {
+    const payments = await db.Payment.findAll({
+      where: { confirmed: false },
+    });
 
-  let block = bitcoin.Block.fromHex(message.toString("hex"));
-  block.transactions.map(tx => {
-    let hash = reverse(tx.getHash()).toString("hex");
-    if (payments.find(p => p.hash === hash)) queue[hash] = 1;
-  });
-});
+    let block = bitcoin.Block.fromHex(message.toString("hex"));
+    block.transactions.map((tx) => {
+      let hash = reverse(tx.getHash()).toString("hex");
+      if (payments.find((p) => p.hash === hash)) queue[hash] = 1;
+    });
+  }
+})();
 
 setInterval(async () => {
   try {
@@ -144,25 +148,25 @@ setInterval(async () => {
       const hash = arr[i];
 
       let account, address, user, total;
-      await db.transaction(async transaction => {
+      await db.transaction(async (transaction) => {
         let p = await db.Payment.findOne({
           where: { hash, confirmed: 0, received: 1 },
           include: [
             {
               model: db.Account,
-              as: "account"
+              as: "account",
             },
             {
               model: db.Invoice,
-              as: "invoice"
+              as: "invoice",
             },
             {
               model: db.User,
-              as: "user"
-            }
+              as: "user",
+            },
           ],
           lock: transaction.LOCK.UPDATE,
-          transaction
+          transaction,
         });
 
         if (p && p.address) address = p.address;
@@ -211,7 +215,7 @@ setInterval(async () => {
           address: c.address,
           amount: total - 100,
           user,
-          limit: total
+          limit: total,
         });
       }
     }