Merge pull request #38 from rsksmart/feat/prisma-txPending-txpool

feat: added txPending and txpool converters, insert, update and delete

src/converters/txPending.converters.js

@@ -0,0 +1,34 @@
+function rawTxPendingToEntity (data) {
+  return {
+    hash: data.hash,
+    blockHash: data.blockHash,
+    from: data.from,
+    to: data.to,
+    blockNumber: data.blockNumber,
+    transactionIndex: data.transactionIndex,
+    nonce: data.nonce,
+    gas: data.gas,
+    gasPrice: data.gasPrice,
+    value: data.value,
+    input: data.input,
+    status: data.status
+  }
+}
+
+function rawTxInPoolToEntity (data) {
+  return {
+    ...rawTxPendingToEntity(data),
+    poolId: data.poolId
+  }
+}
+
+function rawTxPoolToEntity (data) {
+  return {
+    blockNumber: data.blockNumber,
+    pending: data.pending,
+    queued: data.queued,
+    timestamp: String(data.timestamp)
+  }
+}
+
+export {rawTxPendingToEntity, rawTxPoolToEntity, rawTxInPoolToEntity}

src/repositories/txPending.repository.js

@@ -1,39 +1,35 @@
 import { prismaClient } from '../lib/Setup'
+import {rawTxPendingToEntity, rawTxInPoolToEntity} from '../converters/txPending.converters'
+import { mongoQueryToPrisma } from './utils'
 
 export const txPendingRepository = {
-  findOne (query = {}, project = {}, collection) {
-    return collection.findOne(query, project)
-  },
-  find (query = {}, project = {}, collection, sort = {}, limit = 0, isArray = true) {
-    if (isArray) {
-      return collection
-        .find(query, project)
-        .sort(sort)
-        .limit(limit)
-        .toArray()
-    } else {
-      return collection
-        .find(query, project)
-        .sort(sort)
-        .limit(limit)
-    }
-  },
-  countDocuments (query = {}, collection) {
-    return collection.countDocuments(query)
+  async findOne (query = {}, project = {}, collection) {
+    const txPending = await prismaClient.transaction_pending.findFirst({where: mongoQueryToPrisma(query)})
+
+    return txPending
   },
-  aggregate (aggregate, collection) {
-    return collection.aggregate(aggregate).toArray()
+  async find (query = {}, project = {}, collection, sort = {}, limit = 0, isArray = true) {
+    const txsPending = await prismaClient.transaction_pending.findMany({where: mongoQueryToPrisma(query)})
+
+    return txsPending
   },
   async deleteOne (query, collection) {
     await prismaClient.transaction_pending.deleteMany({where: query})
 
-    const mongoRes = collection.deleteOne(query)
+    const mongoRes = await collection.deleteOne(query)
     return mongoRes
   },
-  updateOne (filter, update, options = {}, collection) {
-    return collection.updateOne(filter, update, options)
-  },
-  insertOne (data, collection) {
-    return collection.insertOne(data)
+  async updateOne (filter, update, options = {}, collection) {
+    const {$set: data, poolId} = update
+    const newTxPending = rawTxPendingToEntity(data)
+
+    await prismaClient.$transaction([
+      prismaClient.transaction_pending.upsert({where: filter, update: newTxPending, create: newTxPending}),
+      prismaClient.transaction_in_pool.create({data: rawTxInPoolToEntity({...data, poolId})})
+    ])
+
+    delete update.poolId
+    const mongoRes = await collection.updateOne(filter, update, options)
+    return mongoRes
   }
 }

src/repositories/txPool.repository.js

@@ -1,28 +1,51 @@
+import { prismaClient } from '../lib/Setup'
+import {rawTxPoolToEntity} from '../converters/txPending.converters'
+import {mongoQueryToPrisma} from './utils'
+
 export const txPoolRepository = {
-  findOne (query = {}, project = {}, collection) {
-    return collection.findOne(query, project)
+  async findOne (query = {}, project = {}, collection) {
+    const txPool = await prismaClient.txpool.findFirst({
+      where: mongoQueryToPrisma(query),
+      orderBy: {id: 'desc'},
+      include: {transaction_in_pool: true}
+    })
+
+    txPool.txs = [...txPool.transaction_in_pool]
+    delete txPool.transaction_in_pool
+
+    txPool._id = txPool.id
+    delete txPool.id
+
+    txPool.timestamp = Number(txPool.timestamp)
+
+    return txPool
   },
-  find (query = {}, project = {}, collection, sort = {}, limit = 0, isArray = true) {
-    if (isArray) {
-      return collection
-        .find(query, project)
-        .sort(sort)
-        .limit(limit)
-        .toArray()
-    } else {
-      return collection
-        .find(query, project)
-        .sort(sort)
-        .limit(limit)
-    }
+  async find (query = {}, project = {}, collection, sort = {}, limit = 0, isArray = true) {
+    let txPools = await prismaClient.txpool.findMany({
+      where: query,
+      orderBy: {timestamp: 'desc'},
+      include: {transaction_in_pool: true},
+      take: limit
+    })
+
+    txPools = txPools.map(txPool => {
+      txPool.txs = [...txPool.transaction_in_pool]
+      delete txPool.transaction_in_pool
+
+      txPool._id = txPool.id
+      delete txPool.id
+
+      txPool.timestamp = Number(txPool.timestamp)
+
+      return txPool
+    })
+
+    return txPools
   },
-  countDocuments (query = {}, collection) {
-    return collection.countDocuments(query)
-  },
-  aggregate (aggregate, collection) {
-    return collection.aggregate(aggregate).toArray()
-  },
-  insertOne (data, collection) {
-    return collection.insertOne(data)
+  async insertOne (data, collection) {
+    const txpool = await prismaClient.txpool.create({data: rawTxPoolToEntity(data)})
+
+    const mongoRes = await collection.insertOne(data)
+    return {...mongoRes, id: txpool.id}
   }
 }

src/repositories/utils.js

@@ -36,4 +36,24 @@ function createPrismaSelect (project) {
   return (Object.keys(select).length !== 0) ? select : null
 }
 
-export {mongoSortToPrisma, createPrismaOrderBy, createPrismaSelect}
+// TODO: finish the mapping of the remaining mongo operators
+function mongoQueryToPrisma (query) {
+  const mongoOperatorToPrisma = {
+    $or: 'OR',
+    $and: 'AND'
+  }
+  const newQuery = {}
+
+  for (const key in query) {
+    const value = query[key]
+    if ((value && Object.keys(value).length > 0 && !Array.isArray(value))) {
+      newQuery[mongoOperatorToPrisma[key] || key] = mongoQueryToPrisma(value)
+    } else {
+      newQuery[mongoOperatorToPrisma[key] || key] = value
+    }
+  }
+
+  return newQuery
+}
+
+export {mongoSortToPrisma, createPrismaOrderBy, createPrismaSelect, mongoQueryToPrisma}

src/services/classes/TxPool.js

@@ -2,7 +2,6 @@ import { BlocksBase } from '../../lib/BlocksBase'
 import { isHexString, base64toHex } from '../../lib/utils'
 import { txPoolRepository } from '../../repositories/txPool.repository'
 import { txPendingRepository } from '../../repositories/txPending.repository'
-
 export class TxPool extends BlocksBase {
   constructor (db, options) {
     super(db, options)
@@ -117,18 +116,19 @@ export class TxPool extends BlocksBase {
   async savePoolToDb (pool) {
     try {
       this.log.debug(`Saving txPool to db`)
-      await txPoolRepository.insertOne(pool, this.TxPool)
-      await this.savePendingTxs(pool.txs)
+      const txpool = await txPoolRepository.insertOne(pool, this.TxPool)
+      await this.savePendingTxs(pool.txs, txpool.id)
     } catch (err) {
       this.log.error(`Error saving txPool: ${err}`)
       return Promise.reject(err)
     }
   }
 
-  async savePendingTxs (txs) {
+  async savePendingTxs (txs, poolId) {
     try {
       txs = txs || []
-      await Promise.all(txs.map(tx => txPendingRepository.updateOne({ hash: tx.hash }, { $set: tx }, { upsert: true }, this.PendingTxs)))
+      const savedTxs = await Promise.all(txs.map(tx => txPendingRepository.updateOne({ hash: tx.hash }, { $set: tx, poolId }, { upsert: true }, this.PendingTxs)))
+      return savedTxs
     } catch (err) {
       this.log.error(`Error saving pending transactions: ${err}`)
       return Promise.reject(err)