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

@@ -277,7 +277,7 @@ export const MEMPOOL_PROCESS_DELAY = 100
 // On chronik, the max allowed is 200
 export const CHRONIK_FETCH_N_TXS_PER_PAGE = 200
 
-export const INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY = 16
+export const INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY = 64
 export const TX_EMIT_BATCH_SIZE = 200 // for our generator, not chronik
 export const DB_COMMIT_BATCH_SIZE = 200 // tamanho dos lotes para commit no DB
 export const TX_BATCH_POLLING_DELAY = 500 // delay (ms) between polling for new batches of txs to commit to the DB

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,7 +10,7 @@ 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'
@@ -31,9 +31,11 @@ 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)
+  if (process.env.SKIP_CACHE_REBUILD === undefined) {
+    const uncachedAddressStream = getUserUncachedAddresses(userId)
+    for await (const address of uncachedAddressStream) {
+      void await CacheSet.addressCreation(address)
+    }
   }
   return await getCachedPaymentsForUser(userId)
 }
@@ -153,6 +155,10 @@ export const generateAndCacheGroupedPaymentsAndInfoForAddress = async (address:
     for (const tx of batch) {
       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++
@@ -304,6 +310,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)
@@ -313,10 +320,12 @@ export const initPaymentCache = async (address: Address): Promise<boolean> => {
 }
 
 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)
+  if (process.env.SKIP_CACHE_REBUILD === undefined) {
+    const uncachedAddressStream = getUserUncachedAddresses(userId)
+    for await (const address of uncachedAddressStream) {
+      console.log('[CACHE]: Creating cache for address', address.address)
+      await CacheSet.addressCreation(address)
+    }
   }
   const userButtonIds: string[] = (await fetchPaybuttonArrayByUserId(userId))
     .map(p => p.id)

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

@@ -137,6 +137,7 @@ export class ChronikBlockchainClient {
   mempoolTxsBeingProcessed!: number
 
   private latencyTestFinished: boolean
+  private wsReconnecting = false
 
   constructor (networkSlug: string) {
     this.latencyTestFinished = false
@@ -156,8 +157,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 +182,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 +701,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
     }
   }
 

services/transactionService.ts

@@ -519,21 +519,22 @@ export async function connectTransactionsListToPrices (
       grouped.set(p.timestamp, g)
     }
 
-    // Throw on missing price pairs
     for (const ts of tsArray) {
       const allPrices = grouped.get(ts)
       const formattedDate = moment.unix(ts).format(HUMAN_READABLE_DATE_FORMAT)
 
       if (allPrices == null) {
-        throw new Error(
-          `[PRICES] No price record found for networkId=${networkId} at ${formattedDate}.`
+        console.warn(
+          `[PRICES] No price record found for networkId=${networkId} at ${formattedDate} — skipping affected txs.`
         )
+        continue
       }
 
       if ((allPrices.cad == null) || (allPrices.usd == null)) {
-        throw new Error(
-          `[PRICES] Incomplete price data for networkId=${networkId} at ${formattedDate}. Partial data: ${JSON.stringify(allPrices, null, 2)}`
+        console.warn(
+          `[PRICES] Incomplete price data for networkId=${networkId} at ${formattedDate} — skipping affected txs.`
         )
+        continue
       }
 
       priceByNetworkTs.set(`${networkId}:${ts}`, allPrices as AllPrices)
@@ -544,19 +545,32 @@ export async function connectTransactionsListToPrices (
 
   // Build all join rows (2 per tx: USD + CAD)
   const rows: Prisma.PricesOnTransactionsCreateManyInput[] = []
+  const skippedTxIds: string[] = []
   for (const t of txList) {
     const ts = flattenTimestamp(t.timestamp)
     const allPrices = priceByNetworkTs.get(`${t.address.networkId}:${ts}`)
     if (allPrices == null) {
-      throw new Error(`[PRICES] Missing price pair for networkId ${t.address.networkId} at ${moment.unix(ts).format(HUMAN_READABLE_DATE_FORMAT)}.`)
+      skippedTxIds.push(t.id)
+      continue
     }
     rows.push(...buildPriceTxConnectionInput(t, allPrices))
   }
+
+  if (skippedTxIds.length > 0) {
+    console.warn(`[PRICES] Skipped ${skippedTxIds.length} txs due to missing price data.`)
+  }
+
+  const txIdsToConnect = txList.filter(t => !skippedTxIds.includes(t.id)).map(t => t.id)
+  if (txIdsToConnect.length === 0) {
+    console.warn('[PRICES] No txs to connect after filtering — all had missing prices.')
+    return
+  }
+
   console.log(`[PRICES] Built ${rows.length} price links (2 per tx).`)
 
   await prisma.$transaction(
     async (tx) => {
-      await deletePriceTxConnectionsInChunks(tx, txList.map((t) => t.id))
+      await deletePriceTxConnectionsInChunks(tx, txIdsToConnect)
       await createPriceTxConnectionInChunks(tx, rows)
     },
     { timeout: PRICES_CONNECTION_TIMEOUT }