From bcf89f097f9a62a2aeac70da17ab527b243e811a Mon Sep 17 00:00:00 2001 From: Adam Soltys Date: Mon, 9 Aug 2021 19:11:31 -0700 Subject: back to v5 zmq --- routes/assets.js | 2 +- routes/bitcoin/receive.js | 200 +++++++++++++------------- routes/liquid/receive.js | 353 +++++++++++++++++++++++----------------------- 3 files changed, 272 insertions(+), 283 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) => { + Promise.all( + tx.outs.map(async (o) => { + try { + const { value } = o; + + let address; try { - const { value } = o; + 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); + + let user = await db.User.findOne({ + where: { + username: addresses[address], + }, + }); - if ( - Object.keys(addresses).includes(address) && - !change.includes(address) - ) { - payments.push(hash); - - 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", - }, - }); - - 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 : ""; - - let confirmed = false; - - let { account } = invoice; - - account.pending += value; - await account.save(); - - 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 payment = await db.Payment.create({ - account_id: account.id, + const invoice = await db.Invoice.findOne({ + where: { 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 }); + }, + order: [["id", "DESC"]], + include: { + model: db.Account, + as: "account", + }, + }); + + 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 : ""; - emit(user.username, "account", account); + let confirmed = false; - emit(user.username, "payment", payment); - l.info("bitcoin detected", user.username, value); - notify(user, `${value} SAT payment detected`); - callWebhook(invoice, payment); + let { account } = invoice; + + account.pending += value; + await account.save(); + + 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; } - } catch (e) { - console.log(e); + 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 }); + + 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); + } + }) + ); +}); -(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"); - - 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); - } +zmqRawTx.on("message", async (topic, message, sequence) => { + const hex = message.toString("hex"); - if (payments.includes(blinded.txid)) return; + 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); + } - Promise.all( - tx.vout.map(async (o) => { - try { - if (!(o.scriptPubKey && o.scriptPubKey.addresses)) return; + if (payments.includes(blinded.txid)) return; - const { asset } = o; - const value = toSats(o.value); - const address = o.scriptPubKey.addresses[0]; + Promise.all( + tx.vout.map(async (o) => { + try { + if (!(o.scriptPubKey && o.scriptPubKey.addresses)) return; - if ( - Object.keys(addresses).includes(address) && - !change.includes(address) - ) { - await db.transaction(async (transaction) => { - let user = await getUser(addresses[address], transaction); + const { asset } = o; + const value = toSats(o.value); + const address = o.scriptPubKey.addresses[0]; - let invoice = await db.Invoice.findOne({ - where: { - unconfidential: address, - user_id: user.id, - network: "liquid", - }, - order: [["id", "DESC"]], - }); + if ( + Object.keys(addresses).includes(address) && + !change.includes(address) + ) { + await db.transaction(async (transaction) => { + let user = await getUser(addresses[address], transaction); - if (!invoice) return; + let invoice = await db.Invoice.findOne({ + where: { + unconfidential: address, + user_id: user.id, + network: "liquid", + }, + order: [["id", "DESC"]], + }); - let confirmed = 0; + if (!invoice) return; - 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, - user_id: user.id, - asset, - pubkey: account.pubkey, - pending: value, - index: 0, - }, - transaction - ); - } - - if (config.liquid.walletpass) - await lq.walletPassphrase(config.liquid.walletpass, 300); - - await user.save({ 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 : ""; - - let payment = await db.Payment.create( + let confirmed = 0; + + 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( { - account_id: account.id, + seed: account.seed, + path: account.path, user_id: user.id, - hash: blinded.txid, - amount: value - tip, - currency, - memo, - rate, - received: true, - tip, - confirmed, - address, - network: "liquid", - invoice_id: invoice.id, + asset, + pubkey: account.pubkey, + pending: value, + index: 0, }, - { transaction } + transaction ); + } - payments.push(blinded.txid); - payment = payment.get({ plain: true }); - payment.account = account.get({ plain: true }); + if (config.liquid.walletpass) + await lq.walletPassphrase(config.liquid.walletpass, 300); + + await user.save({ 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 : ""; + + 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 } + ); - 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); + 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); + }); } - }) - ); - } -})(); - -(async () => { - for await (const [topic, message] of zmqRawBlock) { - console.log(message); - try { - const payments = await db.Payment.findAll({ - where: { confirmed: 0 }, - }); + } catch (e) { + l.error("Problem processing transaction", e.message, e.stack); + } + }) + ); +}); - const block = Block.fromHex(message.toString("hex"), true); +zmqRawBlock.on("message", async (topic, message, sequence) => { + try { + const payments = await db.Payment.findAll({ + where: { confirmed: 0 }, + }); - let hash, json; + const block = Block.fromHex(message.toString("hex"), true); - hash = await lq.getBlockHash(block.height); - json = await lq.getBlock(hash, 2); + let hash, json; - 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 user = await getUserById(user_id); + hash = await lq.getBlockHash(block.height); + json = await lq.getBlock(hash, 2); - let account = await db.Account.findOne({ - where: { user_id, asset }, + 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 user = await getUserById(user_id); + + 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 payment = await db.Payment.findOne({ + where: { id: asset_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); + + 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); +}); -- cgit v1.2.3