add more transactions and locks

3 files changed, 136 insertions, 101 deletions

diff --git a/routes/liquid/broadcast.js b/routes/liquid/broadcast.js
index a39349a1..1db166c5 100644
--- a/routes/liquid/broadcast.js
+++ b/routes/liquid/broadcast.js
@@ -4,29 +4,44 @@ module.exports = ah(async (req, res) => {
     let hash = await lq.sendRawTransaction(req.body.tx);
 
     if (user) {
-      let { account } = user;
-      let payment = await db.Payment.create({
-        ...req.body.payment,
-        hash,
-        amount: -req.body.payment.amount,
-        tip: 0,
-        account_id: account.id,
-        user_id: user.id,
-        rate: app.get("rates")[user.currency],
-        currency: user.currency,
-        confirmed: true,
-        received: false,
-        network: "liquid",
-      });
+      await db.transaction(async transaction => {
+        let account = await db.Account.findOne({
+          where: {
+            id: user.account_id
+          },
+          lock: transaction.LOCK.UPDATE,
+          transaction
+        });
+
+        let payment = await db.Payment.create(
+          {
+            ...req.body.payment,
+            hash,
+            amount: -req.body.payment.amount,
+            tip: 0,
+            account_id: account.id,
+            user_id: user.id,
+            rate: app.get("rates")[user.currency],
+            currency: user.currency,
+            confirmed: true,
+            received: false,
+            network: "liquid"
+          },
+          { transaction }
+        );
 
-      account.balance += payment.amount - payment.fee;
-      await account.save();
+        await account.increment(
+          { balance: payment.amount - payment.fee },
+          { transaction }
+        );
+        await account.reload({ transaction });
 
-      payment = payment.get({ plain: true });
-      payment.account = account.get({ plain: true });
-      emit(user.username, "payment", payment);
-      emit(user.username, "account", payment.account);
-      res.send(payment);
+        payment = payment.get({ plain: true });
+        payment.account = account.get({ plain: true });
+        emit(user.username, "payment", payment);
+        emit(user.username, "account", payment.account);
+        res.send(payment);
+      });
     } else {
       res.send(result);
     }
diff --git a/routes/liquid/receive.js b/routes/liquid/receive.js
index 2976bf57..ce1c6d4b 100644
--- a/routes/liquid/receive.js
+++ b/routes/liquid/receive.js
@@ -12,9 +12,11 @@ const network =
     config.liquid.network === "mainnet" ? "liquid" : config.liquid.network
   ];
 
-const getAccount = async params => {
+const getAccount = async (params, transaction) => {
   let account = await db.Account.findOne({
-    where: params
+    where: params,
+    lock: transaction.LOCK.UPDATE,
+    transaction
   });
 
   if (account) {
@@ -39,7 +41,9 @@ const getAccount = async params => {
         pubkey
       },
       order: [["id", "ASC"]],
-      limit: 1
+      limit: 1,
+      lock: transaction.LOCK.UPDATE,
+      transaction
     });
 
     if (existing) {
@@ -51,7 +55,7 @@ const getAccount = async params => {
   params = { ...params, ...{ domain, ticker, precision, name } };
   params.balance = 0;
   params.network = "liquid";
-  return db.Account.create(params);
+  return db.Account.create(params, { transaction });
 };
 
 const zmqRawBlock = zmq.socket("sub");
@@ -82,78 +86,91 @@ zmqRawTx.on("message", async (topic, message, sequence) => {
           Object.keys(addresses).includes(address) &&
           !change.includes(address)
         ) {
-          let user = await getUser(addresses[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"]],
-            include: {
-              model: db.Account,
-              as: "account"
-            }
-          });
-
-          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 } = invoice;
+            let confirmed = 0;
 
-          if (
-            account.asset === asset &&
-            (!account.pubkey || account.network === "liquid")
-          ) {
-            account.pending += value;
-            await account.save();
-          } else {
-            account = await getAccount({
-              seed: account.seed,
-              path: account.path,
-              user_id: user.id,
-              asset,
-              pubkey: account.pubkey,
-              pending: value,
-              index: 0
+            let account = await db.Account.findOne({
+              where: {
+                id: invoice.account_id
+              },
+              lock: transaction.LOCK.UPDATE,
+              transaction
             });
-          }
 
-          if (config.liquid.walletpass)
-            await lq.walletPassphrase(config.liquid.walletpass, 300);
+            console.log(account);
+            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
+              );
+            }
 
-          await user.save();
+            if (config.liquid.walletpass)
+              await lq.walletPassphrase(config.liquid.walletpass, 300);
 
-          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 : "";
+            await user.save({ 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"
-          });
+            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 : "";
 
-          payments.push(blinded.txid);
-          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: blinded.txid,
+                amount: value - tip,
+                currency,
+                memo,
+                rate,
+                received: true,
+                tip,
+                confirmed,
+                address,
+                network: "liquid"
+              },
+              { 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`);
+            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`);
+          });
         }
       } catch (e) {
         l.error("Problem processing transaction", e.message, e.stack);
@@ -279,10 +296,15 @@ setInterval(async () => {
         if (p) {
           let total = p.amount + p.tip;
           p.confirmed = 1;
-          p.account.balance += total;
-          p.account.pending -= Math.min(p.account.pending, total);
-
           await p.account.save({ transaction });
+
+          await p.account.increment({ balance: total }, { transaction });
+          await p.account.decrement(
+            { pending: Math.min(p.account.pending, total) },
+            { transaction }
+          );
+          await p.account.reload({ transaction });
+
           await p.save({ transaction });
 
           p = p.get({ plain: true });
diff --git a/routes/liquid/send.js b/routes/liquid/send.js
index a04bc007..04b90814 100644
--- a/routes/liquid/send.js
+++ b/routes/liquid/send.js
@@ -54,13 +54,16 @@ module.exports = ah(async (req, res) => {
               where: {
                 asset: nonbtc[0],
                 user_id: null
-              }
+              },
+              lock: transaction.LOCK.UPDATE,
+              transaction
             });
 
             if (faucet) {
               covered = faucet.balance;
               if (covered > fee) covered = fee;
-              faucet.balance -= covered;
+              await faucet.decrement({ balance: covered }, { transaction });
+              await faucet.reload({ transaction });
               await faucet.save({ transaction });
             }
 
@@ -88,12 +91,14 @@ module.exports = ah(async (req, res) => {
               balance: account.balance
             });
             throw new Error(
-              `Insufficient funds, need ${total} ${account.ticker === 'BTC' ? 'SAT' : account.ticker}, have ${account.balance}`
+              `Insufficient funds, need ${total} ${
+                account.ticker === "BTC" ? "SAT" : account.ticker
+              }, have ${account.balance}`
             );
           }
 
-          account.balance -= total;
-          await account.save({ transaction });
+          await account.decrement({ balance: total }, { transaction });
+          await account.reload({ transaction });
 
           let payment = {
             amount: -amount,
@@ -113,14 +118,7 @@ module.exports = ah(async (req, res) => {
           payments.push(payment);
         }
       }
-    });
-  } catch (e) {
-    l.error("problem creating liquid payment", user.username, e.message);
-    return res.status(500).send(e.message);
-  }
 
-  try {
-    await db.transaction(async transaction => {
       if (config.liquid.walletpass)
         await lq.walletPassphrase(config.liquid.walletpass, 300);