SHARD-1104 - Postgres DB support

.env.development

@@ -1,4 +1,7 @@
 HOST='127.0.0.1'
+PG_ENABLED=false
+PG_DEFAULT_DB_CONNECTION_STRING='postgresql://localhost/default?user=postgres&password=password'
+PG_SHARDEUM_INDEXER_CONNECTION_STRING='postgresql://localhost/shardeumIndexer?user=postgres&password=password'
 COLLECTOR_MODE='MQ'
 RMQ_HOST='127.0.0.1'
 RMQ_USER='user'

package.json

@@ -55,6 +55,8 @@
     "morgan": "1.9.1",
     "node-cron": "3.0.2",
     "nodemon": "^2.0.20",
+    "pg": "8.12.0",
+    "pg-format": "^1.0.4",
     "point-of-view": "4.6.0",
     "qs": "6.11.0",
     "socket.io": "4.7.1",
@@ -76,6 +78,7 @@
     "@typescript-eslint/eslint-plugin": "5.60.1",
     "@typescript-eslint/parser": "5.60.1",
     "@typescript-eslint/typescript-estree": "5.61.0",
+    "@types/pg-format": "^1.0.5",
     "eslint": "8.44.0",
     "eslint-config-prettier": "8.8.0",
     "eslint-config-standard": "17.1.0",

scripts/repair_account_state.ts

@@ -85,7 +85,7 @@ const saveAccount = async (account: AccountCopy): Promise<void> => {
       config.processData.decodeContractInfo &&
       accountType === AccountType.Account &&
       'account' in accObj.account &&
-      bytesToHex(Uint8Array.from(Object.values(accObj.account.account.codeHash))) !== AccountDB.EOA_CodeHash
+      bytesToHex(Uint8Array.from(Object.values(accObj.account.account?.codeHash || []))) !== AccountDB.EOA_CodeHash
     ) {
       const accountExist = await AccountDB.queryAccountByAccountId(accObj.accountId)
       if (config.verbose) console.log('accountExist', accountExist)

scripts/repair_missing_cycle_block.ts

@@ -3,7 +3,7 @@ dotenv.config()
 
 import * as Crypto from '../src/utils/crypto'
 import * as Storage from '../src/storage'
-import * as db from '../src/storage/sqlite3storage'
+import * as db from '../src/storage/dbStorage'
 import * as CycleDB from '../src/storage/cycle'
 import * as BlockDB from '../src/storage/block'
 import { config, overrideDefaultConfig } from '../src/config'
@@ -65,7 +65,7 @@ async function checkCycleData(startCycleNumber = 0, latestCycleNumber: number):
     const batchSize = 1000
     const cycleBatches: number[][] = []
     let end = startCycleNumber + batchSize
-    for (let start = startCycleNumber; start <= latestCycleNumber; ) {
+    for (let start = startCycleNumber; start <= latestCycleNumber;) {
       if (end > latestCycleNumber) end = latestCycleNumber
       cycleBatches.push(generateNumberArray(start, end))
       start = end + 1
@@ -104,7 +104,7 @@ async function checkBlockData(startBlockNumber = 0, latestBlockNumber: number):
     const batchSize = 1000
     const blockBatches: number[][] = []
     let end = startBlockNumber + batchSize
-    for (let start = startBlockNumber; start <= latestBlockNumber; ) {
+    for (let start = startBlockNumber; start <= latestBlockNumber;) {
       if (end > latestBlockNumber) end = latestBlockNumber
       blockBatches.push(generateNumberArray(start, end))
       start = end + 1

src/cache/LatestBlockCache.ts

@@ -1,6 +1,6 @@
 import { config } from '../config'
 import { blockQueryDelayInMillis } from '../utils/block'
-import * as db from '../storage/sqlite3storage'
+import * as db from '../storage/dbStorage'
 import { DbBlock } from '../types'
 import { sleep } from '../utils'
 
@@ -10,38 +10,36 @@ let getLatestBlockFromDB_running = false
 
 async function getLatestBlockFromDB(): Promise<DbBlock> {
   const before = Date.now()
-  
+
   const delayInMillis = blockQueryDelayInMillis()
-  const sql = `SELECT * FROM (SELECT * FROM blocks ORDER BY number DESC LIMIT 100) AS subquery WHERE timestamp <= ${
-    Date.now() - delayInMillis
-  }`
+  const sql = `SELECT * FROM (SELECT *${config.postgresEnabled ? ', readableBlock::TEXT' : ''} FROM blocks ORDER BY number DESC LIMIT 100) AS subquery WHERE timestamp <= ${Date.now() - delayInMillis}`
 
   //Shouldn't the highest block number always be the latest block?
   //This query seems much faster for the same thing
-  // A:  nope, some reason this only updates once only 10 cycles or so.. even if they come to us out 
+  // A:  nope, some reason this only updates once only 10 cycles or so.. even if they come to us out
   // of order I would think the sort on primary key would still work... oh... it could be because
   // earlier blocks are sent.   I think maybe this also needs the timestamp check,
-  // or we get 10 and then sort through and find one that works 
-  
+  // or we get 10 and then sort through and find one that works
+
   //const sql = `SELECT * FROM blocks ORDER BY number DESC LIMIT 1`
 
   const block: DbBlock = await db.get(sql)
 
   const elapsed = Date.now() - before
-  console.log('SLOW QUERY?  getLatestBlockFromDB   ', elapsed, 'ms' , block.number)
+  console.log('SLOW QUERY?  getLatestBlockFromDB   ', elapsed, 'ms', block.number)
 
   return block
 }
 
 async function updateLatestBlockCacheIfNeeded(): Promise<void> {
   const now = Date.now()
   if (latestBlock === null || now - lastCacheUpdateTimestamp >= config.blockCache.cacheUpdateIntervalInMillis) {
-    try{
-      if(getLatestBlockFromDB_running === true) {
+    try {
+      if (getLatestBlockFromDB_running === true) {
         while (getLatestBlockFromDB_running === true) {
           await sleep(200)
         }
-        console.log('SLOW QUERY?  getLatestBlockFromDB skipped', latestBlock.number )
+        console.log('SLOW QUERY?  getLatestBlockFromDB skipped', latestBlock.number)
         return //other process updated latestBlock
       }
       getLatestBlockFromDB_running = true

src/config/index.ts

@@ -41,6 +41,9 @@ export interface Config {
     port: string
     publicKey: string
   }
+  postgresEnabled: boolean
+  sqliteEnabled: boolean
+  pgDefaultDBConnectionString: string
   rpcUrl: string
   apiUrl: string
   verbose: boolean
@@ -59,6 +62,7 @@ export interface Config {
   findTxHashInOriginalTx: boolean
   enableShardeumIndexer: boolean
   shardeumIndexerSqlitePath: string
+  pgShardeumIndexerDBConnectionString: string
   blockIndexing: {
     enabled: boolean
     blockProductionRate: number
@@ -93,14 +97,17 @@ let config: Config = {
   enableCollectorSocketServer: false,
   port: {
     collector: process.env.COLLECTOR_PORT || '4444',
-    server: process.env.PORT || '6101',
+    server: process.env.PORT || '6100',
     log_server: process.env.LOG_SERVER_PORT || '4446',
   },
   distributorInfo: {
     ip: process.env.DISTRIBUTOR_IP || '127.0.0.1',
     port: process.env.DISTRIBUTOR_PORT || '6100',
     publicKey: '',
   },
+  postgresEnabled: process.env.PG_ENABLED === 'true',
+  sqliteEnabled: !(process.env.PG_ENABLED === 'true'),
+  pgDefaultDBConnectionString: process.env.PG_DEFAULT_DB_CONNECTION_STRING || 'postgresql://localhost/shardeum_default&user=root&password=password',
   rpcUrl: 'http://127.0.0.1:8080',
   apiUrl: '',
   verbose: false,
@@ -117,8 +124,9 @@ let config: Config = {
   },
   enableTxHashCache: false,
   findTxHashInOriginalTx: false,
-  enableShardeumIndexer: true,
+  enableShardeumIndexer: false,
   shardeumIndexerSqlitePath: 'shardeum.sqlite',
+  pgShardeumIndexerDBConnectionString: process.env.PG_SHARDEUM_INDEXER_CONNECTION_STRING || 'postgresql://localhost/shardeum_indexer&user=root&password=password',
   blockIndexing: {
     enabled: true,
     blockProductionRate: 6,

src/server.ts

@@ -40,6 +40,9 @@ import { registerCache } from './cache/LatestBlockCache'
 import { Utils as StringUtils } from '@shardus/types'
 import { healthCheckRouter } from './routes/healthCheck'
 
+import pgFormat from "pg-format"
+
+
 if (config.env == envEnum.DEV) {
   //default debug mode keys
   //  pragma: allowlist nextline secret
@@ -1403,7 +1406,8 @@ const start = async (): Promise<void> => {
         filter.address = StringUtils.safeJsonParse(q.address)
       }
       if (typeof q.address === 'string') {
-        filter.address = [q.address]
+        const address = pgFormat('%s', q.address)
+        filter.address = [address]
       }
 
       if (isValidJson(q.topics) && Array.isArray(StringUtils.safeJsonParse(q.topics))) {