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

constants/index.ts

@@ -269,13 +269,18 @@ 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 = 20
 export const CHRONIK_INITIALIZATION_DELAY = 2000
 export const MEMPOOL_PROCESS_DELAY = 100
+export const CONFIRMED_TX_PROCESS_DELAY = 100
 
-// Number of tries before failing a chronik call (min 1)
-export const CHRONIK_TRIES = 3
+export const CHRONIK_WS_MAX_CONNECTION_ATTEMPTS = 10
+export const CHRONIK_WS_BASE_DELAY_MS = 5000
+
+// Number of tries before failing a chronik HTTP call (min 1)
+export const CHRONIK_HTTP_MAX_TRIES = 3
 // Initial delay between retries in milliseconds. This is multiplied by 2 for each retry.
-export const CHRONIK_RETRY_DELAY_MS = 1000
+export const CHRONIK_HTTP_BASE_DELAY_MS = 1000
 
 /* WIP RENAME ALL THOSE */
 // When fetching some address transactions, number of transactions to fetch at a time.

pages/api/payments/count/index.ts

@@ -1,4 +1,3 @@
-import { CacheGet } from 'redis/index'
 import { fetchUserProfileFromId } from 'services/userService'
 import { setSession } from 'utils/setSession'
 import { getFilteredTransactionCount } from 'services/transactionService'
@@ -28,15 +27,8 @@ export default async (req: any, res: any): Promise<void> => {
     if (typeof req.query.endDate === 'string' && req.query.endDate !== '') {
       endDate = req.query.endDate as string
     }
-    if (((buttonIds !== undefined) && buttonIds.length > 0) ||
-        ((years !== undefined) && years.length > 0) ||
-        (startDate !== undefined && endDate !== undefined && startDate !== '' && endDate !== '')) {
-      const totalCount = await getFilteredTransactionCount(userId, buttonIds, years, timezone, startDate, endDate)
-      res.status(200).json(totalCount)
-    } else {
-      const totalCount = await CacheGet.paymentsCount(userId, timezone)
-      res.status(200).json(totalCount)
-    }
+    const totalCount = await getFilteredTransactionCount(userId, buttonIds, years, timezone, startDate, endDate)
+    res.status(200).json(totalCount)
   } else {
     res.status(405).json({ error: 'Method not allowed' })
   }

pages/dashboard/dashboard.module.css

@@ -250,4 +250,14 @@ body[data-theme='dark'] .show_filters_button img {
   display: flex;
   align-items: center;
   gap: 10px;
+}
+
+.cache_banner {
+  background-color: #fff3cd;
+  color: #856404;
+  padding: 10px 16px;
+  border-radius: 6px;
+  margin-top: 12px;
+  font-size: 14px;
+  font-weight: 500;
 }

pages/dashboard/index.tsx

@@ -140,6 +140,21 @@ export default function Dashboard ({ user }: PaybuttonsProps): React.ReactElemen
     }
   }, [selectedButtonIds])
 
+  useEffect(() => {
+    if (dashboardData?.cacheRebuilding !== true) return
+    const timer = setTimeout(() => {
+      let url = 'api/dashboard'
+      if (selectedButtonIds.length > 0) {
+        url += `?buttonIds=${selectedButtonIds.join(',')}`
+      }
+      fetch(url, { headers: { Timezone: moment.tz.guess() } })
+        .then(async res => await res.json())
+        .then(json => { setDashboardData(json) })
+        .catch(console.error)
+    }, 15000)
+    return () => clearTimeout(timer)
+  }, [dashboardData?.cacheRebuilding, selectedButtonIds])
+
   useEffect(() => {
     setPeriodFromString(dashboardData, activePeriodString)
     if (dashboardData !== undefined) {
@@ -203,6 +218,11 @@ export default function Dashboard ({ user }: PaybuttonsProps): React.ReactElemen
               </div>
             </div>
       )}
+      {dashboardData.cacheRebuilding === true && (
+        <div className={style.cache_banner}>
+          Data still being calculated. Values will update automatically.
+        </div>
+      )}
       <div className={style.number_ctn}>
         <NumberBlock value={'$'.concat(formatQuoteValue(activePeriod.totalRevenue, user.userProfile.preferredCurrencyId)) } text='Revenue' />
         <NumberBlock value={activePeriod.totalPayments} text='Payments' />

redis/dashboardCache.ts

@@ -1,5 +1,5 @@
 import { redis } from './clientInstance'
-import { getPaymentStream } from 'redis/paymentCache'
+import { getPaymentStream, isBackgroundRebuildActive } from 'redis/paymentCache'
 import { ChartData, DashboardData, Payment, ButtonData, PaymentDataByButton, ChartColor, PeriodData, ButtonDisplayData } from './types'
 import { Prisma } from '@prisma/client'
 import moment, { DurationInputArg2 } from 'moment'
@@ -379,7 +379,13 @@ export const getUserDashboardData = async function (userId: string, timezone: st
       timezone,
       paybuttonIds
     )
-    await cacheDashboardData(userId, dashboardData)
+
+    const rebuilding = isBackgroundRebuildActive(userId)
+    if (rebuilding) {
+      dashboardData.cacheRebuilding = true
+    } else {
+      await cacheDashboardData(userId, dashboardData)
+    }
     return dashboardData
   }
   return dashboardData

redis/index.ts

@@ -76,34 +76,34 @@ export const CacheSet = {
 
 type MethodName = 'dashboardData' | 'paymentList' | 'addressBalance' | 'paymentsCount'
 
-interface PendingCalls {
-  [userId: string]: Set<MethodName>
-}
-
 export class CacheGet {
-  private static pendingCalls: PendingCalls = {}
+  private static readonly pendingPromises = new Map<string, Map<MethodName, Promise<any>>>()
 
   private static async executeCall<T>(
     userId: string,
     methodName: MethodName,
     fn: () => Promise<T>
   ): Promise<T> {
-    if (this.pendingCalls[userId] === undefined) {
-      this.pendingCalls[userId] = new Set()
+    let userMap = this.pendingPromises.get(userId)
+    if (userMap === undefined) {
+      userMap = new Map()
+      this.pendingPromises.set(userId, userMap)
     }
 
-    if (this.pendingCalls[userId].has(methodName)) {
-      throw new Error(`Method "${methodName}" is already being executed for user "${userId}".`)
+    const existing = userMap.get(methodName)
+    if (existing !== undefined) {
+      return await existing as T
     }
 
-    this.pendingCalls[userId].add(methodName)
+    const promise = fn()
+    userMap.set(methodName, promise)
 
     try {
-      return await fn()
+      return await promise
     } finally {
-      this.pendingCalls[userId].delete(methodName)
-      if (this.pendingCalls[userId].size === 0) {
-        this.pendingCalls[userId] = undefined as unknown as Set<MethodName>
+      userMap.delete(methodName)
+      if (userMap.size === 0) {
+        this.pendingPromises.delete(userId)
       }
     }
   }

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)}`
@@ -30,13 +30,23 @@ export async function * getUserUncachedAddresses (userId: string): AsyncGenerato
   }
 }
 
-export const getPaymentList = async (userId: string): Promise<Payment[]> => {
-  if (process.env.SKIP_CACHE_REBUILD === undefined) {
-    const uncachedAddressStream = getUserUncachedAddresses(userId)
-    for await (const address of uncachedAddressStream) {
-      void await CacheSet.addressCreation(address)
+const triggerBackgroundRebuildIfNeeded = async (userId: string): Promise<void> => {
+  if (process.env.SKIP_CACHE_REBUILD !== undefined) return
+  const uncachedAddresses: Address[] = []
+  const uncachedAddressStream = getUserUncachedAddresses(userId)
+  for await (const address of uncachedAddressStream) {
+    uncachedAddresses.push(address)
+  }
+  if (uncachedAddresses.length > 0) {
+    if (!isBackgroundRebuildActive(userId)) {
+      console.log(`[CACHE] ${uncachedAddresses.length} uncached addresses for user ${userId}, starting background rebuild`)
     }
+    void cacheAddressesInBackground(uncachedAddresses, userId)
   }
+}
+
+export const getPaymentList = async (userId: string): Promise<Payment[]> => {
+  await triggerBackgroundRebuildIfNeeded(userId)
   return await getCachedPaymentsForUser(userId)
 }
 
@@ -147,20 +157,31 @@ 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++
       }
     }
+    // Throttle batch processing slightly so long cache rebuilds don't monopolize DB resources.
+    await new Promise(resolve => setTimeout(resolve, 200))
   }
+  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
@@ -315,14 +336,43 @@ export const initPaymentCache = async (address: Address): Promise<boolean> => {
   return false
 }
 
-export async function * getPaymentStream (userId: string): AsyncGenerator<Payment> {
-  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 activeBackgroundRebuilds = new Set<string>()
+
+export const isBackgroundRebuildActive = (userId: string): boolean => {
+  return activeBackgroundRebuilds.has(userId)
+}
+
+const cacheAddressesInBackground = async (addresses: Address[], userId: string): Promise<void> => {
+  if (activeBackgroundRebuilds.has(userId)) {
+    console.log(`[CACHE] Background: already running for user ${userId}, skipping`)
+    return
+  }
+  activeBackgroundRebuilds.add(userId)
+  const startTime = Date.now()
+  let cached = 0
+  try {
+    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.slice(0, 20)}...: ${String(err.message ?? err)}`)
+      }
     }
+    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)
+  } finally {
+    activeBackgroundRebuilds.delete(userId)
   }
+}
+
+export async function * getPaymentStream (userId: string): AsyncGenerator<Payment> {
+  await triggerBackgroundRebuildIfNeeded(userId)
+
   const userButtonIds: string[] = (await fetchPaybuttonArrayByUserId(userId))
     .map(p => p.id)
   const weekKeys = await getCachedWeekKeysForUser(userId)
@@ -341,7 +391,7 @@ export async function * getPaymentStream (userId: string): AsyncGenerator<Paymen
         })
 
       for (const payment of weekPayments) {
-        yield payment // Yield one payment at a time
+        yield payment
       }
     }
   }

redis/types.ts

@@ -37,6 +37,7 @@ export interface DashboardData {
     buttons: number
   }
   filtered: boolean
+  cacheRebuilding?: boolean
 }
 
 export interface ButtonDisplayData {

services/addressService.ts

@@ -1,7 +1,6 @@
 import { Prisma, Address } from '@prisma/client'
 import prisma from 'prisma-local/clientInstance'
 import { RESPONSE_MESSAGES } from 'constants/index'
-import { fetchAddressTransactions } from 'services/transactionService'
 import { getNetworkFromSlug } from 'services/networkService'
 
 const addressWithTransactionAndNetwork = Prisma.validator<Prisma.AddressDefaultArgs>()({
@@ -228,17 +227,20 @@ export interface AddressPaymentInfo {
 }
 
 export async function generateAddressPaymentInfo (addressString: string): Promise<AddressPaymentInfo> {
-  const transactionsAmounts = (await fetchAddressTransactions(addressString)).map((t) => t.amount)
-  const balance = transactionsAmounts.reduce((a, b) => {
-    return a.plus(b)
-  }, new Prisma.Decimal(0))
-  const zero = new Prisma.Decimal(0)
-  const paymentCount = transactionsAmounts.filter(t => t > zero).length
-  const info = {
-    balance,
+  const address = await fetchAddressBySubstring(addressString)
+  const [balanceResult, paymentCount] = await Promise.all([
+    prisma.transaction.aggregate({
+      where: { addressId: address.id },
+      _sum: { amount: true }
+    }),
+    prisma.transaction.count({
+      where: { addressId: address.id, amount: { gt: 0 } }
+    })
+  ])
+  return {
+    balance: balanceResult._sum.amount ?? new Prisma.Decimal(0),
     paymentCount
   }
-  return info
 }
 export async function getEarliestUnconfirmedTxTimestampForAddress (addressId: string): Promise<number | undefined> {
   const tx = await prisma.transaction.findFirst({