Migrate vault entries to new schema (#2092)

* Migrate existing vault entries to new schema

* Read+write new vault schema

* Drop VaultEntry table

* Refactor vaultPrismaFragments

* Remove wrong comment

* Remove TODO

* Fix possible race condition on update of vault key

* Remove lib/object.js

Author: ekzyis

api/resolvers/vault.js

@@ -1,42 +1,61 @@
 import { E_VAULT_KEY_EXISTS, GqlAuthenticationError, GqlInputError } from '@/lib/error'
+import { getWalletByType } from '@/wallets/common'
+import { deleteVault, hasVault, vaultNewSchematoTypedef, vaultPrismaFragments } from '@/wallets/vault'
 
 export default {
   Query: {
     getVaultEntries: async (parent, args, { me, models }) => {
       if (!me) throw new GqlAuthenticationError()
 
-      return await models.vaultEntry.findMany({ where: { userId: me.id } })
+      const wallets = await models.wallet.findMany({
+        where: { userId: me.id },
+        include: vaultPrismaFragments.include()
+      })
+
+      const vaultEntries = []
+      for (const wallet of wallets) {
+        vaultEntries.push(...vaultNewSchematoTypedef(wallet).vaultEntries)
+      }
+
+      return vaultEntries
     }
   },
   Mutation: {
     // atomic vault migration
     updateVaultKey: async (parent, { entries, hash }, { me, models }) => {
       if (!me) throw new GqlAuthenticationError()
       if (!hash) throw new GqlInputError('hash required')
-      const txs = []
 
       const { vaultKeyHash: oldKeyHash } = await models.user.findUnique({ where: { id: me.id } })
       if (oldKeyHash) {
-        if (oldKeyHash !== hash) {
-          throw new GqlInputError('vault key already set', E_VAULT_KEY_EXISTS)
-        } else {
+        if (oldKeyHash === hash) {
           return true
         }
-      } else {
-        txs.push(models.user.update({
-          where: { id: me.id },
-          data: { vaultKeyHash: hash }
-        }))
+        throw new GqlInputError('vault key already set', E_VAULT_KEY_EXISTS)
       }
 
-      for (const entry of entries) {
-        txs.push(models.vaultEntry.update({
-          where: { userId_key: { userId: me.id, key: entry.key } },
-          data: { value: entry.value, iv: entry.iv }
-        }))
-      }
-      await models.$transaction(txs)
-      return true
+      return await models.$transaction(async tx => {
+        const wallets = await tx.wallet.findMany({ where: { userId: me.id } })
+        for (const wallet of wallets) {
+          const def = getWalletByType(wallet.type)
+          await tx.wallet.update({
+            where: { id: wallet.id },
+            data: {
+              [def.walletField]: {
+                update: vaultPrismaFragments.upsert({ ...wallet, vaultEntries: entries })
+              }
+            }
+          })
+        }
+
+        // optimistic concurrency control: make sure the user's vault key didn't change while we were updating the wallets
+        await tx.user.update({
+          where: { id: me.id, vaultKeyHash: oldKeyHash },
+          data: { vaultKeyHash: hash }
+        })
+
+        return true
+      })
     },
     clearVault: async (parent, args, { me, models }) => {
       if (!me) throw new GqlAuthenticationError()
@@ -45,7 +64,10 @@ export default {
         where: { id: me.id },
         data: { vaultKeyHash: '' }
       }))
-      txs.push(models.vaultEntry.deleteMany({ where: { userId: me.id } }))
+
+      const wallets = await models.wallet.findMany({ where: { userId: me.id } })
+      txs.push(...wallets.filter(hasVault).map(wallet => deleteVault(models, wallet)))
+
       await models.$transaction(txs)
       return true
     }

api/resolvers/wallet.js

@@ -28,6 +28,7 @@ import { canReceive, getWalletByType } from '@/wallets/common'
 import performPaidAction from '../paidAction'
 import performPayingAction from '../payingAction'
 import { timeoutSignal, withTimeout } from '@/lib/time'
+import { deleteVault, hasVault, vaultNewSchematoTypedef, vaultPrismaFragments } from '@/wallets/vault'
 
 function injectResolvers (resolvers) {
   console.group('injected GraphQL resolvers:')
@@ -42,11 +43,13 @@ function injectResolvers (resolvers) {
         // this mutation was sent from an unsynced client
         // to pass validation, we need to add the existing vault entries for validation
         // in case the client is removing the receiving config
-        existingVaultEntries = await models.vaultEntry.findMany({
+        const wallet = await models.wallet.findUnique({
           where: {
-            walletId: Number(data.id)
-          }
+            id: Number(data.id)
+          },
+          include: vaultPrismaFragments.include()
         })
+        existingVaultEntries = vaultNewSchematoTypedef(wallet).vaultEntries
       }
 
       const validData = await validateWallet(walletDef,
@@ -158,17 +161,17 @@ const resolvers = {
         throw new GqlAuthenticationError()
       }
 
-      return await models.wallet.findMany({
-        include: {
-          vaultEntries: true
-        },
+      const wallets = await models.wallet.findMany({
+        include: vaultPrismaFragments.include(),
         where: {
           userId: me.id
         },
         orderBy: {
           priority: 'asc'
         }
       })
+
+      return wallets.map(vaultNewSchematoTypedef)
     },
     withdrawl: getWithdrawl,
     direct: async (parent, { id }, { me, models }) => {
@@ -554,7 +557,11 @@ const resolvers = {
       }
 
       const logger = walletLogger({ wallet, models })
-      await models.wallet.delete({ where: { userId: me.id, id: Number(id) } })
+
+      await models.$transaction([
+        hasVault(wallet) ? deleteVault(models, wallet) : null,
+        models.wallet.delete({ where: { userId: me.id, id: Number(id) } })
+      ].filter(Boolean))
 
       if (canReceive({ def: getWalletByType(wallet.type), config: wallet.wallet })) {
         logger.info('details for receiving deleted')
@@ -823,78 +830,41 @@ async function upsertWallet (
 
   const txs = []
 
-  if (id) {
-    const oldVaultEntries = await models.vaultEntry.findMany({ where: { userId: me.id, walletId: Number(id) } })
+  const walletWithVault = { ...wallet, vaultEntries }
 
-    // createMany is the set difference of the new - old
-    // deleteMany is the set difference of the old - new
-    // updateMany is the intersection of the old and new
-    const difference = (a = [], b = [], key = 'key') => a.filter(x => !b.find(y => y[key] === x[key]))
-    const intersectionMerge = (a = [], b = [], key = 'key') => a.filter(x => b.find(y => y[key] === x[key]))
-      .map(x => ({ [key]: x[key], ...b.find(y => y[key] === x[key]) }))
+  if (id) {
+    const dbWallet = await models.wallet.findUnique({
+      where: { id: Number(id), userId: me.id }
+    })
 
     txs.push(
       models.wallet.update({
         where: { id: Number(id), userId: me.id },
         data: {
           enabled,
           priority,
-          // client only wallets have no receive config and thus don't have their own table
-          ...(Object.keys(recvConfig).length > 0
-            ? {
-                [wallet.field]: {
-                  upsert: {
-                    create: recvConfig,
-                    update: recvConfig
-                  }
-                }
-              }
-            : {}),
-          ...(vaultEntries
-            ? {
-                vaultEntries: {
-                  deleteMany: difference(oldVaultEntries, vaultEntries, 'key').map(({ key }) => ({
-                    userId: me.id, key
-                  })),
-                  create: difference(vaultEntries, oldVaultEntries, 'key').map(({ key, iv, value }) => ({
-                    key, iv, value, userId: me.id
-                  })),
-                  update: intersectionMerge(oldVaultEntries, vaultEntries, 'key').map(({ key, iv, value }) => ({
-                    where: { userId_key: { userId: me.id, key } },
-                    data: { value, iv }
-                  }))
-                }
-              }
-            : {})
-
+          [wallet.field]: {
+            upsert: {
+              create: { ...recvConfig, ...vaultPrismaFragments.create(walletWithVault) },
+              update: { ...recvConfig, ...vaultPrismaFragments.upsert(walletWithVault) }
+            },
+            // XXX the check is required because the update would fail if there is no row to delete ...
+            update: hasVault(dbWallet) ? vaultPrismaFragments.deleteMissing(walletWithVault) : undefined
+          }
         },
-        include: {
-          vaultEntries: true
-        }
+        include: vaultPrismaFragments.include(walletWithVault)
       })
     )
   } else {
     txs.push(
       models.wallet.create({
-        include: {
-          vaultEntries: true
-        },
+        include: vaultPrismaFragments.include(walletWithVault),
         data: {
           enabled,
           priority,
           userId: me.id,
           type: wallet.type,
-          // client only wallets have no receive config and thus don't have their own table
-          ...(Object.keys(recvConfig).length > 0 ? { [wallet.field]: { create: recvConfig } } : {}),
-          ...(vaultEntries
-            ? {
-                vaultEntries: {
-                  createMany: {
-                    data: vaultEntries?.map(({ key, iv, value }) => ({ key, iv, value, userId: me.id }))
-                  }
-                }
-              }
-            : {})
+          [wallet.field]: { create: { ...recvConfig, ...vaultPrismaFragments.create(walletWithVault) } }
         }
       })
     )
@@ -931,7 +901,7 @@ async function upsertWallet (
   }
 
   const [upsertedWallet] = await models.$transaction(txs)
-  return upsertedWallet
+  return vaultNewSchematoTypedef(upsertedWallet)
 }
 
 export async function createWithdrawal (parent, { invoice, maxFee }, { me, models, lnd, headers, wallet, logger }) {