fix: tx prices linking, dashboard load, initial sync errors

.env.from-dump

@@ -1,2 +1,4 @@
 IGNORE_SEEDING=true
 DONT_EXECUTE_TRIGGERS=true
+FROM_DUMP=true
+SKIP_CACHE_REBUILD=true

.gitignore

@@ -39,3 +39,4 @@ prisma-local/seeds/productionTxs.csv
 paybutton-config.json
 
 dump.sql*
+.forever

Makefile

@@ -37,6 +37,18 @@ dev-from-dump:
 	$(touch_local_env)
 	docker compose -f docker-compose-from-dump.yml --env-file .env --env-file .env.local --env-file .env.from-dump up --build -d
 
+stop-dev-from-dump:
+	docker compose -f docker-compose-from-dump.yml --env-file .env --env-file .env.local --env-file .env.from-dump down
+
+reset-dev-from-dump:
+	make stop-dev-from-dump && make dev-from-dump
+
+reset-dev-from-dump-keep-db:
+	docker compose -f docker-compose-from-dump.yml --env-file .env --env-file .env.local --env-file .env.from-dump up --build -d --force-recreate --no-deps paybutton
+
+nuke-dev-from-dump:
+	docker compose -f docker-compose-from-dump.yml --env-file .env --env-file .env.local --env-file .env.from-dump down -v
+
 logs-dev:
 	docker logs -f paybutton-dev
 

constants/index.ts

@@ -269,8 +269,10 @@ export const XEC_TX_EXPLORER_URL = 'https://explorer.e.cash/tx/'
 export const BCH_TX_EXPLORER_URL = 'https://blockchair.com/bitcoin-cash/transaction/'
 
 export const MAX_MEMPOOL_TXS_TO_PROCESS_AT_A_TIME = 2
+export const MAX_CONFIRMED_TXS_TO_PROCESS_AT_A_TIME = 6
 export const CHRONIK_INITIALIZATION_DELAY = 2000
 export const MEMPOOL_PROCESS_DELAY = 100
+export const CONFIRMED_TX_PROCESS_DELAY = 100
 
 /* WIP RENAME ALL THOSE */
 // When fetching some address transactions, number of transactions to fetch at a time.

docker-compose-from-dump.yml

@@ -42,6 +42,7 @@ services:
       - ./scripts/db/init-supertokens-db.sql:/home/mysql/raw_entrypoint/3-init-supertokens-db.sql
       - ./scripts/db/prisma-shadow-db-fix.sql:/home/mysql/raw_entrypoint/4-prisma-shadow-db-fix.sql
       - ./dump.sql:/home/mysql/raw_entrypoint/5-dump.sql
+      - paybutton-dump-db-data:/var/lib/mysql
 
   users-service:
     container_name: paybutton-users-service
@@ -69,3 +70,6 @@ services:
       - ./redis:/data/redis:z
       - ./scripts:/data/scripts
     command: sh -c "sed -i 's/\\r//g' ./scripts/redis-start.sh && sh ./scripts/redis-start.sh"
+
+volumes:
+  paybutton-dump-db-data:

redis/paymentCache.ts

@@ -10,11 +10,11 @@ import {
 import { fetchAllUserAddresses, AddressPaymentInfo } from 'services/addressService'
 import { fetchPaybuttonArrayByUserId } from 'services/paybuttonService'
 
-import { RESPONSE_MESSAGES, PAYMENT_WEEK_KEY_FORMAT, KeyValueT } from 'constants/index'
+import { RESPONSE_MESSAGES, PAYMENT_WEEK_KEY_FORMAT, KeyValueT, N_OF_QUOTES } from 'constants/index'
 import moment from 'moment-timezone'
 import { CacheSet } from 'redis/index'
 import { ButtonDisplayData, Payment } from './types'
-import { getUserDashboardData } from './dashboardCache'
+import { getUserDashboardData, clearDashboardCache } from './dashboardCache'
 // ADDRESS:payments:YYYY:MM
 const getPaymentsWeekKey = (addressString: string, timestamp: number): string => {
   return `${addressString}:payments:${moment.unix(timestamp).format(PAYMENT_WEEK_KEY_FORMAT)}`
@@ -31,10 +31,20 @@ export async function * getUserUncachedAddresses (userId: string): AsyncGenerato
 }
 
 export const getPaymentList = async (userId: string): Promise<Payment[]> => {
-  const uncachedAddressStream = getUserUncachedAddresses(userId)
-  for await (const address of uncachedAddressStream) {
-    void await CacheSet.addressCreation(address)
+  const allAddresses = await fetchAllUserAddresses(userId, false, false) as Address[]
+  const uncachedAddresses: Address[] = []
+  for (const address of allAddresses) {
+    const keys = await getCachedWeekKeysForAddress(address.address)
+    if (keys.length === 0) {
+      uncachedAddresses.push(address)
+    }
+  }
+
+  if (uncachedAddresses.length > 0 && process.env.SKIP_CACHE_REBUILD === undefined) {
+    console.log(`[CACHE] getPaymentList: ${uncachedAddresses.length} uncached addresses for user ${userId}, starting background rebuild`)
+    void cacheAddressesInBackground(uncachedAddresses, userId)
   }
+
   return await getCachedPaymentsForUser(userId)
 }
 
@@ -145,20 +155,29 @@ export const generatePaymentFromTxWithInvoices = (tx: TransactionWithAddressAndP
 }
 
 export const generateAndCacheGroupedPaymentsAndInfoForAddress = async (address: Address): Promise<GroupedPaymentsAndInfoObject> => {
+  const startTime = Date.now()
   let paymentList: Payment[] = []
   let balance = new Prisma.Decimal(0)
   let paymentCount = 0
+  let txCount = 0
   const txsWithPaybuttonsGenerator = generateTransactionsWithPaybuttonsAndPricesForAddress(address.id)
   for await (const batch of txsWithPaybuttonsGenerator) {
     for (const tx of batch) {
+      txCount++
       balance = balance.plus(tx.amount)
       if (tx.amount.gt(0)) {
+        if (tx.prices.length !== N_OF_QUOTES) {
+          console.warn(`[CACHE] Skipping tx ${tx.hash} — missing price links (${tx.prices.length}/${N_OF_QUOTES})`)
+          continue
+        }
         const payment = generatePaymentFromTx(tx)
         paymentList.push(payment)
         paymentCount++
       }
     }
   }
+  const elapsed = ((Date.now() - startTime) / 1000).toFixed(1)
+  console.log(`[CACHE] Cached address ${address.address.slice(0, 20)}...: ${paymentCount} payments, ${txCount} txs in ${elapsed}s`)
   const info: AddressPaymentInfo = {
     balance,
     paymentCount
@@ -304,6 +323,7 @@ export const clearPaymentCacheForAddress = async (addressString: string): Promis
 }
 
 export const initPaymentCache = async (address: Address): Promise<boolean> => {
+  if (process.env.SKIP_CACHE_REBUILD !== undefined) return false
   const cachedKeys = await getCachedWeekKeysForAddress(address.address)
   if (cachedKeys.length === 0) {
     await CacheSet.addressCreation(address)
@@ -312,12 +332,40 @@ export const initPaymentCache = async (address: Address): Promise<boolean> => {
   return false
 }
 
+const cacheAddressesInBackground = async (addresses: Address[], userId: string): Promise<void> => {
+  const startTime = Date.now()
+  let cached = 0
+  for (const address of addresses) {
+    try {
+      await CacheSet.addressCreation(address)
+      cached++
+      if (cached % 10 === 0 || cached === addresses.length) {
+        console.log(`[CACHE] Background: cached ${cached}/${addresses.length} addresses for user ${userId}`)
+      }
+    } catch (err: any) {
+      console.error(`[CACHE] Background: failed to cache ${address.address}: ${err.message as string}`)
+    }
+  }
+  const elapsed = ((Date.now() - startTime) / 1000).toFixed(1)
+  console.log(`[CACHE] Background: completed ${cached}/${addresses.length} addresses for user ${userId} in ${elapsed}s`)
+  await clearDashboardCache(userId)
+}
+
 export async function * getPaymentStream (userId: string): AsyncGenerator<Payment> {
-  const uncachedAddressStream = getUserUncachedAddresses(userId)
-  for await (const address of uncachedAddressStream) {
-    console.log('[CACHE]: Creating cache for address', address.address)
-    await CacheSet.addressCreation(address)
+  const allAddresses = await fetchAllUserAddresses(userId, false, false) as Address[]
+  const uncachedAddresses: Address[] = []
+  for (const address of allAddresses) {
+    const keys = await getCachedWeekKeysForAddress(address.address)
+    if (keys.length === 0) {
+      uncachedAddresses.push(address)
+    }
   }
+
+  if (uncachedAddresses.length > 0 && process.env.SKIP_CACHE_REBUILD === undefined) {
+    console.log(`[CACHE] ${uncachedAddresses.length}/${allAddresses.length} uncached addresses for user ${userId}, starting background rebuild`)
+    void cacheAddressesInBackground(uncachedAddresses, userId)
+  }
+
   const userButtonIds: string[] = (await fetchPaybuttonArrayByUserId(userId))
     .map(p => p.id)
   const weekKeys = await getCachedWeekKeysForUser(userId)
@@ -336,7 +384,7 @@ export async function * getPaymentStream (userId: string): AsyncGenerator<Paymen
         })
 
       for (const payment of weekPayments) {
-        yield payment // Yield one payment at a time
+        yield payment
       }
     }
   }

scripts/db-entrypoint.sh

@@ -10,11 +10,25 @@ sed_vars () {
 cd /home/mysql/raw_entrypoint || exit
 
 for file in *.sql; do
-    sed_vars "$file"
+    if [[ "$file" == *dump* ]]; then
+      echo "Linking dump file: $file"
+      ln -s /home/mysql/raw_entrypoint/"$file" /home/mysql/entrypoint/"$file"
+    else
+        echo "Copying file $file after seding vars"
+        sed_vars "$file"
+    fi
 done
 
 cd ../entrypoint/ || exit
 
 for file in *.sql; do
-    mariadb -u root -p"$MYSQL_ROOT_PASSWORD" < "$file"
+    if [[ "$file" == *dump* ]]; then
+        filesize=$(stat -Lc%s "$file" 2>/dev/null || stat -Lf%z "$file" 2>/dev/null)
+        echo "Importing $file ($(numfmt --to=iec $filesize)) ..."
+        pv "$file" | mariadb -u root -p"$MYSQL_ROOT_PASSWORD"
+    else
+        echo "Importing $file ..."
+        mariadb -u root -p"$MYSQL_ROOT_PASSWORD" < "$file"
+    fi
+    echo "Done importing $file"
 done

scripts/db/Dockerfile

@@ -3,7 +3,7 @@ FROM mariadb:latest
 RUN mkhomedir_helper mysql
 
 RUN apt-get update || echo && \
-      apt-get install -y gettext || echo
+      apt-get install -y gettext pv || echo
 
 RUN mkdir -p /home/mysql/raw_entrypoint && \
     mkdir -p /home/mysql/entrypoint

scripts/paybutton-server-start.sh

@@ -17,17 +17,19 @@ logtime=$(date +%Y-%m-%d@%H:%M)
 [ -e logs/jobs.log ] && mv logs/jobs.log logs/history/jobs_"$logtime".log
 [ -e logs/ws-server.log ] && mv logs/ws-server.log logs/history/ws-server_"$logtime".log
 
-if [ "$ENVIRONMENT" = "production" ]; then
+if [ "$FROM_DUMP" = "true" ] || [ "$ENVIRONMENT" = "production" ]; then
     yarn prisma migrate deploy || exit 1
     yarn prisma generate || exit 1
-    pm2 start yarn --time --interpreter ash --name jobs --output logs/jobs.log --error logs/jobs.log -- initJobs || exit 1
-    pm2 start yarn --time --interpreter ash --name WSServer --output logs/ws-server.log --error logs/ws-server.log -- initWSServer || exit 1
-    pm2 start yarn --time --interpreter ash --name next --output logs/next.log --error logs/next.log -- prod || exit 1
 else
     yarn prisma migrate dev || exit 1
     yarn prisma db seed || exit 1
-    pm2 start yarn --time --interpreter ash --name jobs --output logs/jobs.log --error logs/jobs.log -- initJobs || exit 1
-    pm2 start yarn --time --interpreter ash --name WSServer --output logs/ws-server.log --error logs/ws-server.log -- initWSServer || exit 1
+fi
+
+pm2 start yarn --time --interpreter ash --name jobs --output logs/jobs.log --error logs/jobs.log -- initJobs || exit 1
+pm2 start yarn --time --interpreter ash --name WSServer --output logs/ws-server.log --error logs/ws-server.log -- initWSServer || exit 1
+if [ "$ENVIRONMENT" = "production" ]; then
+    pm2 start yarn --time --interpreter ash --name next --output logs/next.log --error logs/next.log -- prod || exit 1
+else
     pm2 start yarn --time --interpreter ash --name next --output logs/next.log --error logs/next.log -- dev || exit 1
 fi
 pm2 logs next

services/chronikService.ts

@@ -1,7 +1,7 @@
 import { BlockInfo, ChronikClient, ConnectionStrategy, ScriptUtxo, Tx, WsConfig, WsEndpoint, WsMsgClient, WsSubScriptClient } from 'chronik-client'
 import { encodeCashAddress, decodeCashAddress } from 'ecashaddrjs'
 import { AddressWithTransaction, BlockchainInfo, TransactionDetails, ProcessedMessages, SubbedAddressesLog, SyncAndSubscriptionReturn, SubscriptionReturn, SimpleBlockInfo } from 'types/chronikTypes'
-import { CHRONIK_MESSAGE_CACHE_DELAY, RESPONSE_MESSAGES, XEC_TIMESTAMP_THRESHOLD, XEC_NETWORK_ID, BCH_NETWORK_ID, BCH_TIMESTAMP_THRESHOLD, CHRONIK_FETCH_N_TXS_PER_PAGE, KeyValueT, NETWORK_IDS_FROM_SLUGS, SOCKET_MESSAGES, NETWORK_IDS, NETWORK_TICKERS, MainNetworkSlugsType, MAX_MEMPOOL_TXS_TO_PROCESS_AT_A_TIME, MEMPOOL_PROCESS_DELAY, CHRONIK_INITIALIZATION_DELAY, LATENCY_TEST_CHECK_DELAY, INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY, TX_EMIT_BATCH_SIZE, DB_COMMIT_BATCH_SIZE, MAX_TXS_PER_ADDRESS, TX_BATCH_POLLING_DELAY } from 'constants/index'
+import { CHRONIK_MESSAGE_CACHE_DELAY, RESPONSE_MESSAGES, XEC_TIMESTAMP_THRESHOLD, XEC_NETWORK_ID, BCH_NETWORK_ID, BCH_TIMESTAMP_THRESHOLD, CHRONIK_FETCH_N_TXS_PER_PAGE, KeyValueT, NETWORK_IDS_FROM_SLUGS, SOCKET_MESSAGES, NETWORK_IDS, NETWORK_TICKERS, MainNetworkSlugsType, MAX_MEMPOOL_TXS_TO_PROCESS_AT_A_TIME, MAX_CONFIRMED_TXS_TO_PROCESS_AT_A_TIME, MEMPOOL_PROCESS_DELAY, CONFIRMED_TX_PROCESS_DELAY, CHRONIK_INITIALIZATION_DELAY, LATENCY_TEST_CHECK_DELAY, INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY, TX_EMIT_BATCH_SIZE, DB_COMMIT_BATCH_SIZE, MAX_TXS_PER_ADDRESS, TX_BATCH_POLLING_DELAY } from 'constants/index'
 import { productionAddresses } from 'prisma-local/seeds/addresses'
 import prisma from 'prisma-local/clientInstance'
 import {
@@ -137,6 +137,8 @@ export class ChronikBlockchainClient {
   mempoolTxsBeingProcessed!: number
 
   private latencyTestFinished: boolean
+  private wsReconnecting = false
+  private confirmedTxsBeingProcessed = 0
 
   constructor (networkSlug: string) {
     this.latencyTestFinished = false
@@ -156,8 +158,8 @@ export class ChronikBlockchainClient {
       this.latencyTestFinished = true
       this.chronikWSEndpoint = this.chronik.ws(this.getWsConfig())
       this.confirmedTxsHashesFromLastBlock = []
-      void this.chronikWSEndpoint.waitForOpen()
       this.chronikWSEndpoint.subscribeToBlocks()
+      void this.connectWsWithRetry()
       this.lastProcessedMessages = { confirmed: {}, unconfirmed: {} }
       this.CHRONIK_MSG_PREFIX = `[CHRONIK — ${networkSlug}]`
       this.wsEndpoint = io(`${config.wsBaseURL}/broadcast`, {
@@ -181,6 +183,42 @@ export class ChronikBlockchainClient {
     }
   }
 
+  private async connectWsWithRetry (maxRetries = 10, baseDelay = 5000): Promise<void> {
+    for (let attempt = 1; attempt <= maxRetries; attempt++) {
+      try {
+        await this.chronikWSEndpoint.waitForOpen()
+        console.log(`${this.CHRONIK_MSG_PREFIX}: WebSocket connected.`)
+        return
+      } catch (err: any) {
+        console.error(`${this.CHRONIK_MSG_PREFIX}: WebSocket connection attempt ${attempt}/${maxRetries} failed: ${err.message as string}`)
+        if (attempt < maxRetries) {
+          const delay = Math.min(baseDelay * Math.pow(2, attempt - 1), 60000)
+          console.log(`${this.CHRONIK_MSG_PREFIX}: Retrying WebSocket in ${delay / 1000}s...`)
+          await new Promise(resolve => setTimeout(resolve, delay))
+        }
+      }
+    }
+    console.error(`${this.CHRONIK_MSG_PREFIX}: WebSocket failed after ${maxRetries} attempts. Continuing without real-time updates.`)
+  }
+
+  private async reconnectWs (): Promise<void> {
+    if (this.wsReconnecting) return
+    this.wsReconnecting = true
+    try {
+      const addresses = this.getSubscribedAddresses()
+      this.chronikWSEndpoint = this.chronik.ws(this.getWsConfig())
+      this.chronikWSEndpoint.subscribeToBlocks()
+      for (const addr of addresses) {
+        this.chronikWSEndpoint.subscribeToAddress(addr)
+      }
+      await this.connectWsWithRetry()
+    } catch (err: any) {
+      console.error(`${this.CHRONIK_MSG_PREFIX}: WebSocket reconnection error: ${err.message as string}`)
+    } finally {
+      this.wsReconnecting = false
+    }
+  }
+
   public getUrls (): string[] {
     return this.chronik.proxyInterface().getEndpointArray().map(e => e.url)
   }
@@ -664,10 +702,16 @@ export class ChronikBlockchainClient {
     return {
       onMessage: (msg: WsMsgClient) => { void this.processWsMessage(msg) },
       onError: (e: ws.ErrorEvent) => { console.log(`${this.CHRONIK_MSG_PREFIX}: Chronik webSocket error, type: ${e.type} | message: ${e.message} | error: ${e.error as string}`) },
-      onReconnect: (_: ws.Event) => { console.log(`${this.CHRONIK_MSG_PREFIX}: Chronik webSocket unexpectedly closed.`) },
+      onReconnect: (_: ws.Event) => {
+        console.log(`${this.CHRONIK_MSG_PREFIX}: Chronik webSocket unexpectedly closed. Attempting reconnection...`)
+        void this.reconnectWs()
+      },
       onConnect: (_: ws.Event) => { console.log(`${this.CHRONIK_MSG_PREFIX}: Chronik webSocket connection (re)established.`) },
-      onEnd: (e: ws.Event) => { console.log(`${this.CHRONIK_MSG_PREFIX}: Chronik WebSocket ended, type: ${e.type}.`) },
-      autoReconnect: true
+      onEnd: (e: ws.Event) => {
+        console.log(`${this.CHRONIK_MSG_PREFIX}: Chronik WebSocket ended, type: ${e.type}. Attempting reconnection...`)
+        void this.reconnectWs()
+      },
+      autoReconnect: false
     }
   }
 
@@ -788,6 +832,12 @@ export class ChronikBlockchainClient {
         }
       } else if (msg.msgType === 'TX_CONFIRMED') {
         if (this.isAlreadyBeingProcessed(msg.txid, true)) return
+
+        while (this.confirmedTxsBeingProcessed >= MAX_CONFIRMED_TXS_TO_PROCESS_AT_A_TIME) {
+          await new Promise(resolve => setTimeout(resolve, CONFIRMED_TX_PROCESS_DELAY))
+        }
+
+        this.confirmedTxsBeingProcessed += 1
         try {
           const transaction = await this.fetchTxWithRetry(msg.txid)
           const addressesWithTransactions = await this.getAddressesForTransaction(transaction)
@@ -808,6 +858,8 @@ export class ChronikBlockchainClient {
             const { [msg.txid]: _, ...rest } = this.lastProcessedMessages.confirmed
             this.lastProcessedMessages.confirmed = rest
           }
+        } finally {
+          this.confirmedTxsBeingProcessed = Math.max(0, this.confirmedTxsBeingProcessed - 1)
         }
       } else if (msg.msgType === 'TX_ADDED_TO_MEMPOOL') {
         if (this.isAlreadyBeingProcessed(msg.txid, false)) return