diff options
| -rw-r--r-- | routes/bitcoin/receive.js | 220 |
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, }); } } |