back to v5 zmq

3 files changed, 261 insertions, 272 deletions

diff --git a/routes/assets.js b/routes/assets.js
index 939eb76c..0a71bbb8 100644
--- a/routes/assets.js
+++ b/routes/assets.js
@@ -157,7 +157,7 @@ app.post(
       const brt = await lq.blindRawTransaction(hex, true, [], false);
       const srt = await lq.signRawTransactionWithWallet(brt);
       const allowed = (await lq.testMempoolAccept([srt.hex]))[0].allowed;
-      if (!allowed) throw new Error("issuance rejected by mempool");
+      if (!allowed) throw new Error("issuance rejected by mempool", srt.hex);
       const txid = await lq.sendRawTransaction(srt.hex);
 
       await db.transaction(async transaction => {
diff --git a/routes/bitcoin/receive.js b/routes/bitcoin/receive.js
index 2e8114dd..603db47b 100644
--- a/routes/bitcoin/receive.js
+++ b/routes/bitcoin/receive.js
@@ -1,15 +1,15 @@
 const reverse = require("buffer-reverse");
-const zmq = require("zeromq");
+const zmq = require("zeromq/v5-compat");
 const { Op } = require("sequelize");
 const { fromBase58 } = require("bip32");
 
 const bitcoin = require("bitcoinjs-lib");
 
-const zmqRawBlock = new zmq.Subscriber();
+const zmqRawBlock = zmq.socket("sub");
 zmqRawBlock.connect(config.bitcoin.zmqrawblock);
 zmqRawBlock.subscribe("rawblock");
 
-const zmqRawTx = new zmq.Subscriber();
+const zmqRawTx = zmq.socket("sub");
 zmqRawTx.connect(config.bitcoin.zmqrawtx);
 zmqRawTx.subscribe("rawtx");
 
@@ -20,117 +20,112 @@ const network =
 
 const queue = {};
 
-(async () => {
-  for await (const [topic, message] of zmqRawTx) {
-    const hex = message.toString("hex");
-    let tx = bitcoin.Transaction.fromHex(message);
+zmqRawTx.on("message", async (topic, message, sequence) => {
+  const hex = message.toString("hex");
+  let tx = bitcoin.Transaction.fromHex(message);
 
-    let hash = reverse(tx.getHash()).toString("hex");
+  let hash = reverse(tx.getHash()).toString("hex");
 
-    if (payments.includes(hash)) return;
+  if (payments.includes(hash)) return;
 
-    Promise.all(
-      tx.outs.map(async (o) => {
-        try {
-          const { value } = o;
+  Promise.all(
+    tx.outs.map(async (o) => {
+      try {
+        const { value } = o;
 
-          let address;
-          try {
-            address = bitcoin.address.fromOutputScript(o.script, network);
-          } catch (e) {
-            return;
-          }
+        let address;
+        try {
+          address = bitcoin.address.fromOutputScript(o.script, network);
+        } catch (e) {
+          return;
+        }
 
-          if (
-            Object.keys(addresses).includes(address) &&
-            !change.includes(address)
-          ) {
-            payments.push(hash);
+        if (
+          Object.keys(addresses).includes(address) &&
+          !change.includes(address)
+        ) {
+          payments.push(hash);
 
-            let user = await db.User.findOne({
-              where: {
-                username: addresses[address],
-              },
-            });
+          let user = await db.User.findOne({
+            where: {
+              username: addresses[address],
+            },
+          });
 
-            const invoice = await db.Invoice.findOne({
-              where: {
-                user_id: user.id,
-                network: "bitcoin",
-              },
-              order: [["id", "DESC"]],
-              include: {
-                model: db.Account,
-                as: "account",
-              },
-            });
+          const invoice = await db.Invoice.findOne({
+            where: {
+              user_id: user.id,
+              network: "bitcoin",
+            },
+            order: [["id", "DESC"]],
+            include: {
+              model: db.Account,
+              as: "account",
+            },
+          });
 
-            if (!invoice) return;
+          if (!invoice) return;
 
-            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 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 confirmed = false;
+          let confirmed = false;
 
-            let { account } = invoice;
+          let { account } = invoice;
 
-            account.pending += value;
-            await account.save();
+          account.pending += value;
+          await account.save();
 
-            if (config.bitcoin.walletpass)
-              await bc.walletPassphrase(config.bitcoin.walletpass, 300);
+          if (config.bitcoin.walletpass)
+            await bc.walletPassphrase(config.bitcoin.walletpass, 300);
 
-            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;
+          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;
 
-            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 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, "account", account);
+          emit(user.username, "account", account);
 
-            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, "payment", payment);
+          l.info("bitcoin detected", user.username, value);
+          notify(user, `${value} SAT payment detected`);
+          callWebhook(invoice, payment);
         }
-      })
-    );
-  }
-})();
+      } catch (e) {
+        console.log(e);
+      }
+    })
+  );
+});
 
-(async () => {
-  for await (const [topic, message] of zmqRawBlock) {
+zmqRawBlock.on("message", async (topic, message, sequence) => {
     const payments = await db.Payment.findAll({
       where: { confirmed: false },
     });
@@ -140,8 +135,7 @@ const queue = {};
       let hash = reverse(tx.getHash()).toString("hex");
       if (payments.find((p) => p.hash === hash)) queue[hash] = 1;
     });
-  }
-})();
+  });
 
 setInterval(async () => {
   try {
diff --git a/routes/liquid/receive.js b/routes/liquid/receive.js
index a30ff985..2396755f 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");
+const zmq = require("zeromq/v5-compat");
 const { Op } = require("sequelize");
 const { fromBase58 } = require("bip32");
 const bitcoin = require("bitcoinjs-lib");
@@ -57,180 +57,204 @@ const getAccount = async (params, transaction) => {
   return db.Account.create(params, { transaction });
 };
 
-const zmqRawBlock = new zmq.Subscriber();
+const zmqRawBlock = zmq.socket("sub");
 zmqRawBlock.connect(config.liquid.zmqrawblock);
 zmqRawBlock.subscribe("rawblock");
 
-const zmqRawTx = new zmq.Subscriber();
+const zmqRawTx = zmq.socket("sub");
 zmqRawTx.connect(config.liquid.zmqrawtx);
 zmqRawTx.subscribe("rawtx");
 
 const queue = {};
 
-(async () => {
-  for await (const [topic, message] of zmqRawTx) {
-    const hex = message.toString("hex");
+zmqRawTx.on("message", async (topic, message, sequence) => {
+  const hex = message.toString("hex");
 
-    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);
-    }
+  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;
+  if (payments.includes(blinded.txid)) return;
 
-    Promise.all(
-      tx.vout.map(async (o) => {
-        try {
-          if (!(o.scriptPubKey && o.scriptPubKey.addresses)) return;
+  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];
+        const { asset } = o;
+        const value = toSats(o.value);
+        const address = o.scriptPubKey.addresses[0];
 
-          if (
-            Object.keys(addresses).includes(address) &&
-            !change.includes(address)
-          ) {
-            await db.transaction(async (transaction) => {
-              let user = await getUser(addresses[address], transaction);
+        if (
+          Object.keys(addresses).includes(address) &&
+          !change.includes(address)
+        ) {
+          await db.transaction(async (transaction) => {
+            let user = await getUser(addresses[address], transaction);
+
+            let invoice = await db.Invoice.findOne({
+              where: {
+                unconfidential: address,
+                user_id: user.id,
+                network: "liquid",
+              },
+              order: [["id", "DESC"]],
+            });
+
+            if (!invoice) return;
+
+            let confirmed = 0;
+
+            let account = await db.Account.findOne({
+              where: {
+                id: invoice.account_id,
+              },
+              lock: transaction.LOCK.UPDATE,
+              transaction,
+            });
 
-              let invoice = await db.Invoice.findOne({
-                where: {
-                  unconfidential: address,
+            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,
-                  network: "liquid",
+                  asset,
+                  pubkey: account.pubkey,
+                  pending: value,
+                  index: 0,
                 },
-                order: [["id", "DESC"]],
-              });
+                transaction
+              );
+            }
 
-              if (!invoice) return;
+            if (config.liquid.walletpass)
+              await lq.walletPassphrase(config.liquid.walletpass, 300);
 
-              let confirmed = 0;
+            await user.save({ transaction });
 
-              let account = await db.Account.findOne({
-                where: {
-                  id: invoice.account_id,
-                },
-                lock: transaction.LOCK.UPDATE,
-                transaction,
-              });
+            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 (
-                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
-                );
-              }
+            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 (config.liquid.walletpass)
-                await lq.walletPassphrase(config.liquid.walletpass, 300);
+            payments.push(blinded.txid);
+            payment = payment.get({ plain: true });
+            payment.account = account.get({ plain: true });
 
-              await user.save({ transaction });
+            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);
+      }
+    })
+  );
+});
 
-              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 : "";
+zmqRawBlock.on("message", async (topic, message, sequence) => {
+  try {
+    const payments = await db.Payment.findAll({
+      where: { confirmed: 0 },
+    });
 
-              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 }
-              );
+    const block = Block.fromHex(message.toString("hex"), true);
 
-              payments.push(blinded.txid);
-              payment = payment.get({ plain: true });
-              payment.account = account.get({ plain: true });
+    let hash, json;
 
-              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);
-        }
-      })
-    );
-  }
-})();
+    hash = await lq.getBlockHash(block.height);
+    json = await lq.getBlock(hash, 2);
 
-(async () => {
-  for await (const [topic, message] of zmqRawBlock) {
-    console.log(message);
-    try {
-      const payments = await db.Payment.findAll({
-        where: { confirmed: 0 },
-      });
+    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];
 
-      const block = Block.fromHex(message.toString("hex"), true);
+          const user = await getUserById(user_id);
 
-      let hash, json;
+          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 });
 
-      hash = await lq.getBlockHash(block.height);
-      json = await lq.getBlock(hash, 2);
+          let payment = await db.Payment.findOne({
+            where: { id: asset_payment_id },
+            include: {
+              model: db.Account,
+              as: "account",
+            },
+            lock: transaction.LOCK.UPDATE,
+            transaction,
+          });
 
-      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];
+          payment.confirmed = true;
+          await payment.save({ transaction });
+          payment = payment.get({ plain: true });
+          payment.account = account.get({ plain: true });
 
-            const user = await getUserById(user_id);
+          emit(user.username, "account", account);
+          emit(user.username, "payment", payment);
 
-            let account = await db.Account.findOne({
-              where: { user_id, asset },
+          if (token) {
+            account = await db.Account.findOne({
+              where: { user_id, asset: token },
               lock: transaction.LOCK.UPDATE,
               transaction,
             });
-            account.balance = asset_amount * SATS;
+            account.balance = token_amount * SATS;
             account.pending = 0;
             await account.save({ transaction });
 
-            let payment = await db.Payment.findOne({
-              where: { id: asset_payment_id },
+            payment = await db.Payment.findOne({
+              where: { id: token_payment_id },
               include: {
                 model: db.Account,
                 as: "account",
@@ -241,49 +265,20 @@ const queue = {};
 
             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);
-
-            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);
-    }
+          }
+        });
+      } else if (payments.find((p) => p.hash === tx.txid)) queue[tx.txid] = 1;
+    });
+  } catch (e) {
+    return console.log(e);
   }
-})();
+});
 
 setInterval(async () => {
   try {
@@ -374,4 +369,4 @@ setInterval(async () => {
   } catch (e) {
     l.error("problem processing queued liquid transaction", e.message, e.stack);
   }
-}, 1000);
+});