fix zmq 6 syntax

3 files changed, 192 insertions, 185 deletions

diff --git a/routes/bitcoin/receive.js b/routes/bitcoin/receive.js
index 572987cc..2e8114dd 100644
--- a/routes/bitcoin/receive.js
+++ b/routes/bitcoin/receive.js
@@ -7,7 +7,7 @@ const bitcoin = require("bitcoinjs-lib");
 
 const zmqRawBlock = new zmq.Subscriber();
 zmqRawBlock.connect(config.bitcoin.zmqrawblock);
-zmqRawBlock.subscribe("rawtx");
+zmqRawBlock.subscribe("rawblock");
 
 const zmqRawTx = new zmq.Subscriber();
 zmqRawTx.connect(config.bitcoin.zmqrawtx);
@@ -18,7 +18,7 @@ const network =
     config.bitcoin.network === "mainnet" ? "bitcoin" : config.bitcoin.network
   ];
 
-let queue = {};
+const queue = {};
 
 (async () => {
   for await (const [topic, message] of zmqRawTx) {
@@ -127,7 +127,9 @@ let queue = {};
       })
     );
   }
+})();
 
+(async () => {
   for await (const [topic, message] of zmqRawBlock) {
     const payments = await db.Payment.findAll({
       where: { confirmed: false },
diff --git a/routes/liquid/receive.js b/routes/liquid/receive.js
index 2c88075c..a30ff985 100644
--- a/routes/liquid/receive.js
+++ b/routes/liquid/receive.js
@@ -1,6 +1,6 @@
 const axios = require("axios");
 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");
@@ -15,7 +15,7 @@ const getAccount = async (params, transaction) => {
   let account = await db.Account.findOne({
     where: params,
     lock: transaction.LOCK.UPDATE,
-    transaction
+    transaction,
   });
 
   if (account) {
@@ -37,12 +37,12 @@ const getAccount = async (params, transaction) => {
     const existing = await db.Account.findOne({
       where: {
         asset,
-        pubkey
+        pubkey,
       },
       order: [["id", "ASC"]],
       limit: 1,
       lock: transaction.LOCK.UPDATE,
-      transaction
+      transaction,
     });
 
     if (existing) {
@@ -57,228 +57,233 @@ const getAccount = async (params, transaction) => {
   return db.Account.create(params, { transaction });
 };
 
-const zmqRawBlock = zmq.socket("sub");
+const zmqRawBlock = new zmq.Subscriber();
 zmqRawBlock.connect(config.liquid.zmqrawblock);
 zmqRawBlock.subscribe("rawblock");
 
-const zmqRawTx = zmq.socket("sub");
+const zmqRawTx = new zmq.Subscriber();
 zmqRawTx.connect(config.liquid.zmqrawtx);
 zmqRawTx.subscribe("rawtx");
 
-zmqRawTx.on("message", async (topic, message, sequence) => {
-  const hex = message.toString("hex");
+const queue = {};
 
-  let unblinded, tx, blinded;
-  try {
-    unblinded = await lq.unblindRawTransaction(hex);
-    tx = await lq.decodeRawTransaction(unblinded.hex);
-    blinded = await lq.decodeRawTransaction(hex);
-  } catch (e) {
-    return console.log(e);
-  }
-
-  if (payments.includes(blinded.txid)) return;
+(async () => {
+  for await (const [topic, message] of zmqRawTx) {
+    const hex = message.toString("hex");
 
-  Promise.all(
-    tx.vout.map(async o => {
-      try {
-        if (!(o.scriptPubKey && o.scriptPubKey.addresses)) return;
-
-        const { asset } = o;
-        const value = toSats(o.value);
-        const address = o.scriptPubKey.addresses[0];
+    let unblinded, tx, blinded;
+    try {
+      unblinded = await lq.unblindRawTransaction(hex);
+      tx = await lq.decodeRawTransaction(unblinded.hex);
+      blinded = await lq.decodeRawTransaction(hex);
+    } catch (e) {
+      return console.log(e);
+    }
 
-        if (
-          Object.keys(addresses).includes(address) &&
-          !change.includes(address)
-        ) {
-          await db.transaction(async transaction => {
-            let user = await getUser(addresses[address], transaction);
+    if (payments.includes(blinded.txid)) return;
 
-            let invoice = await db.Invoice.findOne({
-              where: {
-                unconfidential: address,
-                user_id: user.id,
-                network: "liquid"
-              },
-              order: [["id", "DESC"]]
-            });
+    Promise.all(
+      tx.vout.map(async (o) => {
+        try {
+          if (!(o.scriptPubKey && o.scriptPubKey.addresses)) return;
 
-            if (!invoice) return;
+          const { asset } = o;
+          const value = toSats(o.value);
+          const address = o.scriptPubKey.addresses[0];
 
-            let confirmed = 0;
+          if (
+            Object.keys(addresses).includes(address) &&
+            !change.includes(address)
+          ) {
+            await db.transaction(async (transaction) => {
+              let user = await getUser(addresses[address], transaction);
 
-            let account = await db.Account.findOne({
-              where: {
-                id: invoice.account_id
-              },
-              lock: transaction.LOCK.UPDATE,
-              transaction
-            });
-
-            if (
-              account.asset === asset &&
-              (!account.pubkey || account.network === "liquid")
-            ) {
-              await account.increment({ pending: value }, { transaction });
-              await account.reload({ transaction });
-            } else {
-              account = await getAccount(
-                {
-                  seed: account.seed,
-                  path: account.path,
+              let invoice = await db.Invoice.findOne({
+                where: {
+                  unconfidential: address,
                   user_id: user.id,
-                  asset,
-                  pubkey: account.pubkey,
-                  pending: value,
-                  index: 0
+                  network: "liquid",
                 },
-                transaction
-              );
-            }
+                order: [["id", "DESC"]],
+              });
 
-            if (config.liquid.walletpass)
-              await lq.walletPassphrase(config.liquid.walletpass, 300);
+              if (!invoice) return;
 
-            await user.save({ transaction });
+              let confirmed = 0;
 
-            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 = await db.Account.findOne({
+                where: {
+                  id: invoice.account_id,
+                },
+                lock: transaction.LOCK.UPDATE,
+                transaction,
+              });
 
-            let payment = await db.Payment.create(
-              {
-                account_id: account.id,
-                user_id: user.id,
-                hash: blinded.txid,
-                amount: value - tip,
-                currency,
-                memo,
-                rate,
-                received: true,
-                tip,
-                confirmed,
-                address,
-                network: "liquid",
-                invoice_id: invoice.id
-              },
-              { transaction }
-            );
+              if (
+                account.asset === asset &&
+                (!account.pubkey || account.network === "liquid")
+              ) {
+                await account.increment({ pending: value }, { transaction });
+                await account.reload({ transaction });
+              } else {
+                account = await getAccount(
+                  {
+                    seed: account.seed,
+                    path: account.path,
+                    user_id: user.id,
+                    asset,
+                    pubkey: account.pubkey,
+                    pending: value,
+                    index: 0,
+                  },
+                  transaction
+                );
+              }
 
-            payments.push(blinded.txid);
-            payment = payment.get({ plain: true });
-            payment.account = account.get({ plain: true });
-
-            emit(user.username, "payment", payment);
-            emit(user.username, "account", payment.account);
-            l.info("liquid detected", address, user.username, asset, value);
-            notify(user, `${value} SAT payment detected`);
-            callWebhook(invoice, payment);
-          });
-        }
-      } catch (e) {
-        l.error("Problem processing transaction", e.message, e.stack);
-      }
-    })
-  );
-});
+              if (config.liquid.walletpass)
+                await lq.walletPassphrase(config.liquid.walletpass, 300);
 
-let queue = {};
+              await user.save({ transaction });
 
-zmqRawBlock.on("message", async (topic, message, sequence) => {
-  try {
-    const payments = await db.Payment.findAll({
-      where: { confirmed: 0 }
-    });
+              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 : "";
 
-    const block = Block.fromHex(message.toString("hex"), true);
+              let payment = await db.Payment.create(
+                {
+                  account_id: account.id,
+                  user_id: user.id,
+                  hash: blinded.txid,
+                  amount: value - tip,
+                  currency,
+                  memo,
+                  rate,
+                  received: true,
+                  tip,
+                  confirmed,
+                  address,
+                  network: "liquid",
+                  invoice_id: invoice.id,
+                },
+                { transaction }
+              );
 
-    let hash, json;
+              payments.push(blinded.txid);
+              payment = payment.get({ plain: true });
+              payment.account = account.get({ plain: true });
 
-    hash = await lq.getBlockHash(block.height);
-    json = await lq.getBlock(hash, 2);
+              emit(user.username, "payment", payment);
+              emit(user.username, "account", payment.account);
+              l.info("liquid detected", address, user.username, asset, value);
+              notify(user, `${value} SAT payment detected`);
+              callWebhook(invoice, payment);
+            });
+          }
+        } catch (e) {
+          l.error("Problem processing transaction", e.message, e.stack);
+        }
+      })
+    );
+  }
+})();
 
-    json.tx.map(async tx => {
-      if (issuances[tx.txid]) {
-        await db.transaction(async transaction => {
-          const {
-            user_id,
-            asset,
-            asset_amount,
-            asset_payment_id,
-            token,
-            token_amount,
-            token_payment_id
-          } = issuances[tx.txid];
+(async () => {
+  for await (const [topic, message] of zmqRawBlock) {
+    console.log(message);
+    try {
+      const payments = await db.Payment.findAll({
+        where: { confirmed: 0 },
+      });
 
-          const user = await getUserById(user_id);
+      const block = Block.fromHex(message.toString("hex"), true);
 
-          let account = await db.Account.findOne({
-            where: { user_id, asset },
-            lock: transaction.LOCK.UPDATE,
-            transaction
-          });
-          account.balance = asset_amount * SATS;
-          account.pending = 0;
-          await account.save({ transaction });
+      let hash, json;
 
-          let payment = await db.Payment.findOne({
-            where: { id: asset_payment_id },
-            include: {
-              model: db.Account,
-              as: "account"
-            },
-            lock: transaction.LOCK.UPDATE,
-            transaction
-          });
+      hash = await lq.getBlockHash(block.height);
+      json = await lq.getBlock(hash, 2);
 
-          payment.confirmed = true;
-          await payment.save({ transaction });
-          payment = payment.get({ plain: true });
-          payment.account = account.get({ plain: true });
+      json.tx.map(async (tx) => {
+        if (issuances[tx.txid]) {
+          await db.transaction(async (transaction) => {
+            const {
+              user_id,
+              asset,
+              asset_amount,
+              asset_payment_id,
+              token,
+              token_amount,
+              token_payment_id,
+            } = issuances[tx.txid];
 
-          emit(user.username, "account", account);
-          emit(user.username, "payment", payment);
+            const user = await getUserById(user_id);
 
-          if (token) {
-            account = await db.Account.findOne({
-              where: { user_id, asset: token },
+            let account = await db.Account.findOne({
+              where: { user_id, asset },
               lock: transaction.LOCK.UPDATE,
-              transaction
+              transaction,
             });
-            account.balance = token_amount * SATS;
+            account.balance = asset_amount * SATS;
             account.pending = 0;
             await account.save({ transaction });
 
-            payment = await db.Payment.findOne({
-              where: { id: token_payment_id },
+            let payment = await db.Payment.findOne({
+              where: { id: asset_payment_id },
               include: {
                 model: db.Account,
-                as: "account"
+                as: "account",
               },
               lock: transaction.LOCK.UPDATE,
-              transaction
+              transaction,
             });
 
             payment.confirmed = true;
             await payment.save({ transaction });
-
             payment = payment.get({ plain: true });
             payment.account = account.get({ plain: true });
 
             emit(user.username, "account", account);
             emit(user.username, "payment", payment);
-          }
-        });
-      } else if (payments.find(p => p.hash === tx.txid)) queue[tx.txid] = 1;
-    });
-  } catch (e) {
-    return console.log(e);
+
+            if (token) {
+              account = await db.Account.findOne({
+                where: { user_id, asset: token },
+                lock: transaction.LOCK.UPDATE,
+                transaction,
+              });
+              account.balance = token_amount * SATS;
+              account.pending = 0;
+              await account.save({ transaction });
+
+              payment = await db.Payment.findOne({
+                where: { id: token_payment_id },
+                include: {
+                  model: db.Account,
+                  as: "account",
+                },
+                lock: transaction.LOCK.UPDATE,
+                transaction,
+              });
+
+              payment.confirmed = true;
+              await payment.save({ transaction });
+
+              payment = payment.get({ plain: true });
+              payment.account = account.get({ plain: true });
+
+              emit(user.username, "account", account);
+              emit(user.username, "payment", payment);
+            }
+          });
+        } else if (payments.find((p) => p.hash === tx.txid)) queue[tx.txid] = 1;
+      });
+    } catch (e) {
+      return console.log(e);
+    }
   }
-});
+})();
 
 setInterval(async () => {
   try {
@@ -288,25 +293,25 @@ setInterval(async () => {
       const hash = arr[i];
 
       let account, address, user, total, p;
-      await db.transaction(async transaction => {
+      await db.transaction(async (transaction) => {
         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,
         });
 
         ({ account, address, user } = p);
@@ -362,7 +367,7 @@ setInterval(async () => {
           address: c.address,
           amount: total - 100,
           user,
-          limit: total
+          limit: total,
         });
       }
     }
diff --git a/routes/users.js b/routes/users.js
index 23f016c7..24138fb6 100644
--- a/routes/users.js
+++ b/routes/users.js
@@ -5,7 +5,7 @@ const authenticator = require("otplib").authenticator;
 const getAccount = require("../lib/account");
 const Sequelize = require("sequelize");
 const bitcoin = require("bitcoinjs-lib");
-const liquid = require("@asoltys/liquidjs-lib");
+// const liquid = require("@asoltys/liquidjs-lib");
 const { fromBase58, fromPrivateKey } = require("bip32");
 const { Mutex } = require("async-mutex");
 const bip32 = require("bip32");