refactor: cleanup comments

chedieck · chedieck · commit 6332239c449b · 2025-08-25T16:37:46.000-03:00
diff --git a/services/chronikService.ts b/services/chronikService.ts
@@ -288,24 +288,21 @@ export class ChronikBlockchainClient {
     return (await this.chronik.script(type, hash160).history(page, pageSize)).txs
   }
 
-  // For each address, fetch PAGE_CONCURRENCY pages in parallel (“burst”),
-  // then use the burst’s newest/oldest timestamps to decide whether to continue.
-  // Yields happen only in the generator body (after each slice finishes, and at final flush).
+  /*
+   * For each address, fetch PAGE_CONCURRENCY pages in parallel (“burst”),
+   * then use the burst’s newest/oldest timestamps to decide whether to continue.
+   * Yields happen only in the generator body (after each slice finishes, and at final flush).
+  */
   private async * fetchLatestTxsForAddresses (
     addresses: Address[]
   ): AsyncGenerator<Prisma.TransactionUncheckedCreateInput[]> {
-    // 1024 -> 7:10m
-    // 512  -> 7:34.175
-    // 256  -> 7:47.876
-    const pagesPerBurstPerAddress = 1 // pageConcurrency
     const logPrefix = `${this.CHRONIK_MSG_PREFIX}[PARALLEL FETCHING]`
 
     console.log(
       `${logPrefix}: Will fetch latest txs for ${addresses.length} addresses ` +
-      `(addressConcurrency=${INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY}, pageConcurrency=${pagesPerBurstPerAddress}).`
+      `(addressConcurrency=${INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY}, pageConcurrency=1).`
     )
 
-    // Shared accumulation buffer — emitted only at the top-level generator yields
     let transactionsToEmitBuffer: Prisma.TransactionUncheckedCreateInput[] = []
 
     for (let i = 0; i < addresses.length; i += INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY) {
@@ -321,82 +318,59 @@ export class ChronikBlockchainClient {
         let hasReachedStoppingCondition = false
 
         while (!hasReachedStoppingCondition) {
-          const pageIndicesInBurst = Array.from(
-            { length: pagesPerBurstPerAddress },
-            (_, k) => nextBurstBasePageIndex + k
-          )
-
-          // Fetch one "burst" of pages for this address in parallel.
-          // Swallow individual page errors so one failing page doesn't cancel the address worker.
-          const burstFetchResults = await Promise.all(
-            pageIndicesInBurst.map(async (pageIndex) => {
-              try {
-                const value = await this.getPaginatedTxs(address.address, pageIndex, CHRONIK_FETCH_N_TXS_PER_PAGE)
-                return { page: pageIndex, value }
-              } catch (err: any) {
-                console.warn(`${addrLogPrefix} page=${pageIndex} failed: ${err?.message as string ?? err as string}`)
-                return { page: pageIndex, value: [] as any[] }
-              }
-            })
-          )
+          const pageIndex = nextBurstBasePageIndex
+
+          // Fetch the single page for this burst; swallow page errors.
+          let pageTxs: any[] = []
+          try {
+            pageTxs = await this.getPaginatedTxs(address.address, pageIndex, CHRONIK_FETCH_N_TXS_PER_PAGE)
+          } catch (err: any) {
+            console.warn(`${addrLogPrefix} page=${pageIndex} failed: ${err?.message as string ?? err as string}`)
+            pageTxs = []
+          }
 
-          // Only consider non-empty pages for timestamp bounds and processing.
-          const nonEmptyPageResults = burstFetchResults.filter(r => r.value.length > 0)
-          if (nonEmptyPageResults.length === 0) {
+          // If the page is empty, treat as "EMPTY ADDRESS" and stop.
+          if (pageTxs.length === 0) {
             console.log(`${addrLogPrefix} EMPTY ADDRESS`)
-            break // nothing in this burst -> done with this address
+            break
           }
 
-          // Determine burst-level timestamp bounds
-          const newestTimestampAcrossBurst = Math.max(
-            ...nonEmptyPageResults.map(r => Number(r.value[0].block?.timestamp ?? r.value[0].timeFirstSeen))
-          )
-          const oldestTimestampAcrossBurst = Math.min(
-            ...nonEmptyPageResults.map(r => {
-              const pageTxs = r.value
-              const lastTx = pageTxs[pageTxs.length - 1]
-              return Number(lastTx.block?.timestamp ?? lastTx.timeFirstSeen)
-            })
-          )
+          // Burst-level bounds collapse to this single page
+          const newestTimestampAcrossBurst = Number(pageTxs[0].block?.timestamp ?? pageTxs[0].timeFirstSeen)
+          const lastTxInPage = pageTxs[pageTxs.length - 1]
+          const oldestTimestampAcrossBurst = Number(lastTxInPage.block?.timestamp ?? lastTxInPage.timeFirstSeen)
 
-          // If the newest tx across the entire burst is older than our last sync, we can quit immediately.
+          // If even the newest is older than lastSync, stop.
           if (newestTimestampAcrossBurst < lastSyncedTimestampSeconds) {
             console.log(`${addrLogPrefix} NO NEW TXS`)
             break
           }
 
-          // Process each non-empty page in ascending page order.
+          // Process this page
           let keptTransactionsInBurstCount = 0
-          for (const pageResult of nonEmptyPageResults.sort((a, b) => a.page - b.page)) {
-            const pageTxs = pageResult.value
-
-            const filteredTxs = pageTxs
-              .filter(txThresholdFilter)
-              .filter(t => t.block === undefined || t.block.timestamp >= lastSyncedTimestampSeconds)
-
-            if (filteredTxs.length > 0) {
-              const txRowsToCreate = await Promise.all(
-                filteredTxs.map(async t => await this.getTransactionFromChronikTransaction(t, address))
-              )
-              transactionsToEmitBuffer.push(...txRowsToCreate)
-              keptTransactionsInBurstCount += txRowsToCreate.length
-            }
 
-            // If this page’s oldest tx is already older than our last sync, we can stop after the burst.
-            const oldestTimestampInPage = Number(
-              pageTxs[pageTxs.length - 1].block?.timestamp ?? pageTxs[pageTxs.length - 1].timeFirstSeen
+          const filteredTxs = pageTxs
+            .filter(txThresholdFilter)
+            .filter(t => t.block === undefined || t.block.timestamp >= lastSyncedTimestampSeconds)
+
+          if (filteredTxs.length > 0) {
+            const txRowsToCreate = await Promise.all(
+              filteredTxs.map(async t => await this.getTransactionFromChronikTransaction(t, address))
             )
-            if (oldestTimestampInPage < lastSyncedTimestampSeconds) {
-              hasReachedStoppingCondition = true
-              // continue the for-loop to finish logging consistently
-            }
+            transactionsToEmitBuffer.push(...txRowsToCreate)
+            keptTransactionsInBurstCount += txRowsToCreate.length
+          }
+
+          // If the page’s oldest tx is older than lastSync, stop after this burst.
+          const oldestTimestampInPage = oldestTimestampAcrossBurst
+          if (oldestTimestampInPage < lastSyncedTimestampSeconds) {
+            hasReachedStoppingCondition = true
           }
 
           console.log(`${addrLogPrefix} ${keptTransactionsInBurstCount} new txs...`)
 
-          nextBurstBasePageIndex += pagesPerBurstPerAddress
+          nextBurstBasePageIndex += 1
 
-          // Fast-stop: if we kept nothing in this burst and the burst’s oldest is older than lastSync, we’re done.
           if (keptTransactionsInBurstCount === 0 && oldestTimestampAcrossBurst < lastSyncedTimestampSeconds) {
             hasReachedStoppingCondition = true
           }
@@ -719,19 +693,16 @@ export class ChronikBlockchainClient {
     console.time(`${this.CHRONIK_MSG_PREFIX} syncAddresses`)
     await setSyncingBatch(addresses.map(a => a.address), true)
 
-    // per-address counters
     const perAddrCount = new Map<string, number>()
     addresses.forEach(a => perAddrCount.set(a.id, 0))
 
-    // commit buffer
     let toCommit: Prisma.TransactionUncheckedCreateInput[] = []
 
     try {
       const pfx = `${this.CHRONIK_MSG_PREFIX}[PARALLEL FETCHING]`
-      // consume generator: it yields batches of prepared txs
-      console.log(`${pfx} will fetch batches of ${INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY} txs from chronik`)
+      console.log(`${pfx} will fetch batches of ${INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY} addresses from chronik`)
       for await (const batch of this.fetchLatestTxsForAddresses(addresses)) {
-        console.log(`${pfx} fetched batch of ${batch.length} tx from chronik`)
+        console.log(`${pfx} fetched batch of ${batch.length} txs from chronik`)
         // count per address before committing
         for (const tx of batch) {
           perAddrCount.set(tx.addressId, (perAddrCount.get(tx.addressId) ?? 0) + 1)
@@ -753,13 +724,11 @@ export class ChronikBlockchainClient {
 
           // broadcast/triggers after commit
           if (created.length > 0) {
-            // Always broadcast per tx (keeps existing behavior)
             const triggerBatch: BroadcastTxData[] = []
             for (const tx of created) {
               const bd = this.broadcastIncomingTx(tx.address.address, tx, []) // inputAddresses left empty in bulk
               triggerBatch.push(bd)
             }
-            // Then, if enabled, execute triggers **in batch**
             if (runTriggers) {
               console.log(`${this.CHRONIK_MSG_PREFIX} executing trigger batch — broadcasts=${triggerBatch.length}`)
               await executeTriggersBatch(triggerBatch, this.networkId)
diff --git a/services/transactionService.ts b/services/transactionService.ts
@@ -404,7 +404,7 @@ async function createPriceTxConnectionInChunks (
     const slice = rows.slice(i, i + PRICES_CONNECTION_BATCH_SIZE)
     await client.pricesOnTransactions.createMany({
       data: slice,
-      skipDuplicates: true // respects unique(priceId, transactionId); safe like upsert
+      skipDuplicates: true
     })
   }
   console.log('[PRICES] Inserted all price links.')
@@ -428,7 +428,7 @@ export async function connectTransactionsListToPrices (txList: Transaction[]): P
 
   console.log(`[PRICES] Preparing to connect ${txList.length} txs to prices...`)
 
-  // Collect unique (networkId -> Set<flattenedTimestamp>)
+  // collect UNIQUE (networkId, timestamp) pairs
   const networkIdToTimestamps = new Map<number, Set<number>>()
   await Promise.all(txList.map(async (t) => {
     const networkId = await getTransactionNetworkId(t)
@@ -438,13 +438,13 @@ export async function connectTransactionsListToPrices (txList: Transaction[]): P
     networkIdToTimestamps.set(networkId, set)
   }))
 
-  // Fetch AllPrices for each unique (networkId, ts)
+  // fetch AllPrices for each unique (networkId, timestamp)
   const timestampToPrice: Record<number, AllPrices> = {}
   let pairs = 0
-  for (const [networkId, stamps] of networkIdToTimestamps.entries()) {
-    for (const ts of stamps) {
+  for (const [networkId, timestamps] of networkIdToTimestamps.entries()) {
+    for (const ts of timestamps) {
       pairs++
-      // outside any ITX; fine to run in parallel-ish
+      // not parallel
       const allPrices = await fetchPricesForNetworkAndTimestamp(networkId, ts, prisma)
       timestampToPrice[ts] = allPrices
     }
@@ -460,7 +460,6 @@ export async function connectTransactionsListToPrices (txList: Transaction[]): P
   }
   console.log(`[PRICES] Built ${rows.length} price links (2 per tx).`)
 
-  // One deleteMany + chunked createMany inside a single interactive tx
   await prisma.$transaction(async (tx) => {
     console.log(`[PRICES] Disconnecting existing price links for ${txList.length} txs...`)
     await tx.pricesOnTransactions.deleteMany({
diff --git a/services/triggerService.ts b/services/triggerService.ts
@@ -274,16 +274,16 @@ async function fetchTriggersGroupedByAddress (addresses: string[]): Promise<Map<
     include: triggerWithPaybuttonAndUserInclude
   })
 
-  const byAddr = new Map<string, TriggerWithPaybuttonAndUser[]>()
+  const triggersByAddress = new Map<string, TriggerWithPaybuttonAndUser[]>()
   for (const t of triggers as TriggerWithPaybuttonAndUser[]) {
     for (const conn of t.paybutton.addresses) {
       const addr = conn.address.address
-      const arr = byAddr.get(addr) ?? []
+      const arr = triggersByAddress.get(addr) ?? []
       arr.push(t)
-      byAddr.set(addr, arr)
+      triggersByAddress.set(addr, arr)
     }
   }
-  return byAddr
+  return triggersByAddress
 }
 
 function buildPostParams (
@@ -391,7 +391,7 @@ async function persistLogsAndDecrements (
     }
   }
 
-  // 2) Then try credits in a separate transaction
+  // 2) try credits in a separate transaction
   const userIds = Array.from(new Set([
     ...Object.keys(acceptedEmailPerUser),
     ...Object.keys(acceptedPostsPerUser)
@@ -410,16 +410,25 @@ async function persistLogsAndDecrements (
       for (const id of userIds) {
         const row = byId.get(id)
         if (row == null) continue
-        const reqEmail = acceptedEmailPerUser[id] ?? 0
-        const reqPost = acceptedPostsPerUser[id] ?? 0
-        const decEmail = Math.min(Math.max(reqEmail, 0), row.emailCredits ?? 0)
-        const decPost = Math.min(Math.max(reqPost, 0), row.postCredits ?? 0)
-        if (decEmail > 0 || decPost > 0) {
+        const requestedEmailCredits = acceptedEmailPerUser[id] ?? 0
+        const requestedPostCredits = acceptedPostsPerUser[id] ?? 0
+
+        // Clamp to [0, availableCredits]
+        const emailCreditsToDecrement = Math.min(
+          Math.max(requestedEmailCredits, 0),
+          row.emailCredits ?? 0
+        )
+
+        const postCreditsToDecrement = Math.min(
+          Math.max(requestedPostCredits, 0),
+          row.postCredits ?? 0
+        )
+        if (emailCreditsToDecrement > 0 || postCreditsToDecrement > 0) {
           updates.push(tx.userProfile.update({
             where: { id },
             data: {
-              ...(decEmail > 0 ? { emailCredits: { decrement: decEmail } } : {}),
-              ...(decPost > 0 ? { postCredits: { decrement: decPost } } : {})
+              ...(emailCreditsToDecrement > 0 ? { emailCredits: { decrement: emailCreditsToDecrement } } : {}),
+              ...(postCreditsToDecrement > 0 ? { postCredits: { decrement: postCreditsToDecrement } } : {})
             }
           }))
         }
@@ -428,7 +437,7 @@ async function persistLogsAndDecrements (
     })
   } catch (e: any) {
     console.error(`[TRIGGER]: credit decrement tx failed: ${e?.message as string ?? e as string}`)
-    // logs already written; we don’t rollback them
+    // logs already written;
   }
 }
 
@@ -466,20 +475,20 @@ function appendOutOfCreditsLogs (
   queue: TriggerTask[],
   startIndex: number,
   logs: Prisma.TriggerLogCreateManyInput[],
-  kind: 'PostData' | 'SendEmail'
+  triggerType: 'PostData' | 'SendEmail'
 ): void {
   const msg =
-    kind === 'PostData'
+    triggerType === 'PostData'
       ? RESPONSE_MESSAGES.USER_OUT_OF_POST_CREDITS_400.message
       : RESPONSE_MESSAGES.USER_OUT_OF_EMAIL_CREDITS_400.message
 
   for (let i = startIndex; i < queue.length; i++) {
     logs.push({
       triggerId: queue[i].triggerId,
       isError: true,
-      actionType: kind,
+      actionType: triggerType,
       data: JSON.stringify({
-        errorName: kind === 'PostData' ? 'USER_OUT_OF_POST_CREDITS' : 'USER_OUT_OF_EMAIL_CREDITS',
+        errorName: triggerType === 'PostData' ? 'USER_OUT_OF_POST_CREDITS' : 'USER_OUT_OF_EMAIL_CREDITS',
         errorMessage: msg,
         errorStack: ''
       })
@@ -489,7 +498,7 @@ function appendOutOfCreditsLogs (
 
 export async function executeTriggersBatch (broadcasts: BroadcastTxData[], networkId: number): Promise<void> {
   if (process.env.DONT_EXECUTE_TRIGGERS === 'true') {
-    console.log(`DONT_EXECUTE_TRIGGERS in env, skipping batch execution of triggers ${broadcasts.length}`)
+    console.log(`DONT_EXECUTE_TRIGGERS in env, skipping batch execution of ${broadcasts.length} triggers`)
     return
   }
 
@@ -509,7 +518,7 @@ export async function executeTriggersBatch (broadcasts: BroadcastTxData[], netwo
   const logs: Prisma.TriggerLogCreateManyInput[] = []
 
   // Build queues
-  console.log(`[TRIGGER ${currency}]: preparing batch — txItems=${txItems.length} addresses=${uniqueAddresses.length}`)
+  console.log(`[TRIGGER ${currency}]: preparing batch — txs=${txItems.length} addresses=${uniqueAddresses.length}`)
 
   for (const { address, tx } of txItems) {
     const triggers = triggersByAddress.get(address) ?? []
@@ -583,7 +592,7 @@ export async function executeTriggersBatch (broadcasts: BroadcastTxData[], netwo
     }
   }
 
-  // Build accepted maps for decrements (charge only accepted)
+  // count accepted triggers for decrements (charge only accepted)
   const postsAcceptedByUser = Object.fromEntries(postResults.map(r => [r.userId, r.accepted]))
   const emailsAcceptedByUser = Object.fromEntries(emailResults.map(r => [r.userId, r.accepted]))
 
