From de06cbf6141679b2e10ca01bee1e372efc356c91 Mon Sep 17 00:00:00 2001 From: Adam Soltys Date: Thu, 24 Jun 2021 02:52:33 -0400 Subject: only send on-chain after db transaction --- routes/bitcoin/send.js | 55 +++++++++++++++++++++++++------------------------- 1 file changed, 27 insertions(+), 28 deletions(-) (limited to 'routes/bitcoin') diff --git a/routes/bitcoin/send.js b/routes/bitcoin/send.js index cab79e06..470889b3 100644 --- a/routes/bitcoin/send.js +++ b/routes/bitcoin/send.js @@ -1,4 +1,3 @@ -const bitcoin = require("bitcoinjs-lib"); const btc = config.liquid.btcasset; module.exports = ah(async (req, res) => { @@ -32,10 +31,10 @@ module.exports = ah(async (req, res) => { total = total - change + fee; let amount = total - fee; - let params; + let account, params; try { await db.transaction(async transaction => { - let account = await db.Account.findOne({ + account = await db.Account.findOne({ where: { id: user.account_id }, @@ -62,38 +61,38 @@ module.exports = ah(async (req, res) => { await account.decrement({ balance: total }, { transaction }); await account.reload({ transaction }); + }); - params = { - amount: -amount, - fee, - memo, - account_id: account.id, - user_id: user.id, - rate: app.get("rates")[user.currency], - currency: user.currency, - address, - confirmed: true, - received: false, - network: "bitcoin" - }; + params = { + amount: -amount, + fee, + memo, + account_id: account.id, + user_id: user.id, + rate: app.get("rates")[user.currency], + currency: user.currency, + address, + confirmed: true, + received: false, + network: "bitcoin" + }; - if (config.bitcoin.walletpass) - await bc.walletPassphrase(config.bitcoin.walletpass, 300); + if (config.bitcoin.walletpass) + await bc.walletPassphrase(config.bitcoin.walletpass, 300); - hex = (await bc.signRawTransactionWithWallet(hex)).hex; - params.hash = await bc.sendRawTransaction(hex); + hex = (await bc.signRawTransactionWithWallet(hex)).hex; + params.hash = await bc.sendRawTransaction(hex); - let payment = await db.Payment.create(params, { transaction }); + let payment = await db.Payment.create(params); - payment = payment.get({ plain: true }); - payment.account = account.get({ plain: true }); + payment = payment.get({ plain: true }); + payment.account = account.get({ plain: true }); - emit(user.username, "payment", payment); - res.send(payment); + emit(user.username, "payment", payment); + res.send(payment); - payments.push(params.hash); - l.info("sent bitcoin", user.username, total); - }); + payments.push(params.hash); + l.info("sent bitcoin", user.username, total); } catch (e) { l.error("error sending bitcoin", e.message); return res.status(500).send(e.message); -- cgit v1.2.3 From 5858323cf0bb920c9f967239b7a1ef7915cf0824 Mon Sep 17 00:00:00 2001 From: Adam Soltys Date: Wed, 4 Aug 2021 17:08:25 -0700 Subject: don't wait for webhook to complete --- routes/assets.js | 11 ++++++++++- routes/bitcoin/receive.js | 4 ++-- routes/lightning/receive.js | 2 +- routes/liquid/receive.js | 2 +- 4 files changed, 14 insertions(+), 5 deletions(-) (limited to 'routes/bitcoin') diff --git a/routes/assets.js b/routes/assets.js index fe667dd9..939eb76c 100644 --- a/routes/assets.js +++ b/routes/assets.js @@ -111,7 +111,15 @@ app.post( if (filename) contract.filename = filename; - l.info("attempting issuance", req.user.username, contract, "address", address, "pubkey", pubkey); + l.info( + "attempting issuance", + req.user.username, + contract, + "address", + address, + "pubkey", + pubkey + ); sha256.update(JSON.stringify(contract)); const hash = sha256.digest("hex"); @@ -128,6 +136,7 @@ app.post( contract_hash }; + params.asset_amount = asset_amount / (SATS / 10 ** precision); if (token_amount) { diff --git a/routes/bitcoin/receive.js b/routes/bitcoin/receive.js index d782627d..67ccdb45 100644 --- a/routes/bitcoin/receive.js +++ b/routes/bitcoin/receive.js @@ -114,7 +114,7 @@ zmqRawTx.on("message", async (topic, message, sequence) => { emit(user.username, "payment", payment); l.info("bitcoin detected", user.username, value); notify(user, `${value} SAT payment detected`); - await callWebhook(invoice, payment); + callWebhook(invoice, payment); } } catch (e) { console.log(e); @@ -189,7 +189,7 @@ setInterval(async () => { emit(user.username, "payment", p); l.info("bitcoin confirmed", user.username, p.amount, p.tip); notify(user, `${total} SAT payment confirmed`); - await callWebhook(p.invoice, p); + callWebhook(p.invoice, p); } else { l.warn("couldn't find bitcoin payment", hash); } diff --git a/routes/lightning/receive.js b/routes/lightning/receive.js index 8d2aa750..39c5be15 100644 --- a/routes/lightning/receive.js +++ b/routes/lightning/receive.js @@ -67,7 +67,7 @@ const handlePayment = async msg => { payment = payment.get({ plain: true }); payment.account = account.get({ plain: true }); - await callWebhook(invoice, payment); + callWebhook(invoice, payment); emit(user.username, "payment", payment); emit(user.username, "account", payment.account); diff --git a/routes/liquid/receive.js b/routes/liquid/receive.js index 259db7c7..d3d4aec8 100644 --- a/routes/liquid/receive.js +++ b/routes/liquid/receive.js @@ -175,7 +175,7 @@ zmqRawTx.on("message", async (topic, message, sequence) => { emit(user.username, "account", payment.account); l.info("liquid detected", address, user.username, asset, value); notify(user, `${value} SAT payment detected`); - await callWebhook(invoice, payment); + callWebhook(invoice, payment); }); } } catch (e) { -- cgit v1.2.3 From 9b03ed9abead5d8d254fe537a8e003c6e468f63d Mon Sep 17 00:00:00 2001 From: Adam Soltys Date: Fri, 6 Aug 2021 08:51:34 -0700 Subject: zmq 6 syntax --- routes/bitcoin/receive.js | 234 +++++++++++++++++++++++----------------------- 1 file changed, 119 insertions(+), 115 deletions(-) (limited to 'routes/bitcoin') 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); + + 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; - const invoice = await db.Invoice.findOne({ - where: { + let payment = await db.Payment.create({ + account_id: account.id, 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; + 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); } - 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); } - } catch (e) { - console.log(e); - } - }) - ); -}); - -let queue = {}; + }) + ); + } -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, }); } } -- cgit v1.2.3 From 7f427d1c9e229bfd70e913db451a873ce561b8db Mon Sep 17 00:00:00 2001 From: Adam Soltys Date: Fri, 6 Aug 2021 14:58:01 -0700 Subject: fix zmq 6 syntax --- routes/bitcoin/receive.js | 6 +- routes/liquid/receive.js | 383 +++++++++++++++++++++++----------------------- routes/users.js | 2 +- 3 files changed, 199 insertions(+), 192 deletions(-) (limited to 'routes/bitcoin') 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); - } +(async () => { + for await (const [topic, message] of zmqRawTx) { + const hex = message.toString("hex"); - if (payments.includes(blinded.txid)) 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]; - - 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"]] - }); + 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 (!invoice) return; + if (payments.includes(blinded.txid)) return; - let confirmed = 0; + Promise.all( + tx.vout.map(async (o) => { + try { + if (!(o.scriptPubKey && o.scriptPubKey.addresses)) return; - let account = await db.Account.findOne({ - where: { - id: invoice.account_id - }, - lock: transaction.LOCK.UPDATE, - transaction - }); + const { asset } = o; + const value = toSats(o.value); + const address = o.scriptPubKey.addresses[0]; - 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, + 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, - 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); - - 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 } - ); + if (!invoice) return; - 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); - } - }) - ); -}); + let confirmed = 0; -let queue = {}; - -zmqRawBlock.on("message", async (topic, message, sequence) => { - try { - const payments = await db.Payment.findAll({ - where: { confirmed: 0 } - }); - - const block = Block.fromHex(message.toString("hex"), true); - - let hash, json; + 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( + { + 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 } + ); - hash = await lq.getBlockHash(block.height); - json = await lq.getBlock(hash, 2); + payments.push(blinded.txid); + 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, "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); + } + }) + ); + } +})(); + +(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"); -- cgit v1.2.3 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(-) (limited to 'routes/bitcoin') 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