aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--routes/assets.js2
-rw-r--r--routes/bitcoin/receive.js188
-rw-r--r--routes/liquid/receive.js343
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);
+});
generated by cgit v1.2.3 (git 2.51.0) at 2026-10-15 20:54:26 +0000