wip2

chedieck · chedieck · commit 76269c4966ef · 2025-09-12T21:21:13.000-03:00
diff --git a/services/chronikService.ts b/services/chronikService.ts
@@ -106,7 +106,7 @@ export function getNullDataScriptData (outputScript: string): OpReturnData | nul
   return ret
 }
 
-type ChronikTxWithAddress = { tx: Tx; address: Address }
+interface ChronikTxWithAddress { tx: Tx, address: Address }
 interface FetchedTxsBatch {
   chronikTxs: ChronikTxWithAddress[]
   addressesSynced: string[]
@@ -310,8 +310,8 @@ export class ChronikBlockchainClient {
     )
 
     let chronikTxs: ChronikTxWithAddress[] = []
-
     let lastBatchAddresses: string[] = []
+
     for (let i = 0; i < addresses.length; i += INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY) {
       const addressBatchSlice = addresses.slice(i, i + INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY)
       lastBatchAddresses = addressBatchSlice.map(a => a.address)
@@ -328,89 +328,69 @@ export class ChronikBlockchainClient {
 
         while (!hasReachedStoppingCondition) {
           const pageIndex = nextBurstBasePageIndex
-
-          // Fetch the single page for this burst; swallow page errors.
           let pageTxs: Tx[] = []
+
           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}`)
+            console.warn(`${addrLogPrefix} page=${pageIndex} failed: ${err.message as string}`)
             pageTxs = []
           }
 
-          // If the page is empty, treat as "EMPTY ADDRESS" and stop.
           if (pageTxs.length === 0) {
             console.log(`${addrLogPrefix} EMPTY ADDRESS`)
             break
           }
 
-          // 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)
+          const newestTs = Number(pageTxs[0].block?.timestamp ?? pageTxs[0].timeFirstSeen)
+          const oldestTs = Number(pageTxs[pageTxs.length - 1].block?.timestamp ?? pageTxs[pageTxs.length - 1].timeFirstSeen)
 
-          // If even the newest is older than lastSync, stop.
-          if (newestTimestampAcrossBurst < lastSyncedTimestampSeconds) {
+          if (newestTs < lastSyncedTimestampSeconds) {
             console.log(`${addrLogPrefix} NO NEW TXS`)
             break
           }
 
-          // Process this page
-          let keptTransactionsInBurstCount = 0
-
           pageTxs = pageTxs
             .filter(txThresholdFilter)
             .filter(t => t.block === undefined || t.block.timestamp >= lastSyncedTimestampSeconds)
 
           if (pageTxs.length > 0) {
-            chronikTxs.push(...pageTxs.map(tx => ({ tx, address: address })))
-            keptTransactionsInBurstCount += pageTxs.length
+            chronikTxs.push(...pageTxs.map(tx => ({ tx, address })))
           }
 
-          // If the page’s oldest tx is older than lastSync, stop after this burst.
-          const oldestTimestampInPage = oldestTimestampAcrossBurst
-          if (oldestTimestampInPage < lastSyncedTimestampSeconds) {
+          if (oldestTs < lastSyncedTimestampSeconds) {
             hasReachedStoppingCondition = true
           }
 
-          console.log(`${addrLogPrefix} ${keptTransactionsInBurstCount} new txs...`)
-
           nextBurstBasePageIndex += 1
-
-          if (keptTransactionsInBurstCount === 0 && oldestTimestampAcrossBurst < lastSyncedTimestampSeconds) {
+          if (pageTxs.length === 0 && oldestTs < lastSyncedTimestampSeconds) {
             hasReachedStoppingCondition = true
           }
         }
       })
 
-      // Wait for this slice of address workers; continue even if some workers throw.
       await Promise.all(
         perAddressWorkers.map(async worker =>
-          await worker.catch(err => {
-            console.error(`${logPrefix}: address job failed: ${err?.message as string ?? err as string}`)
-          })
+          await worker.catch(err => console.error(`${logPrefix}: address job failed: ${err.message as string}`))
         )
       )
 
-      // Emit any full chunks we’ve accumulated after this slice
+      // Emit full chunks of chronik txs (addressesSynced vazio)
       while (chronikTxs.length >= TX_EMIT_BATCH_SIZE) {
         const chronikTxsSlice = chronikTxs.slice(0, TX_EMIT_BATCH_SIZE)
         chronikTxs = chronikTxs.slice(TX_EMIT_BATCH_SIZE)
-        yield {
-          chronikTxs: chronikTxsSlice,
-          addressesSynced: lastBatchAddresses
-        }
+        yield { chronikTxs: chronikTxsSlice, addressesSynced: [] }
       }
+
+      // Emit marcador para este slice (sem txs, só addressesSynced)
+      yield { chronikTxs: [], addressesSynced: lastBatchAddresses }
     }
 
-    // Final flush
+    // Final flush de txs (addressesSynced vazio)
     if (chronikTxs.length > 0) {
       const remaining = chronikTxs
       chronikTxs = []
-      yield {
-        chronikTxs: remaining,
-        addressesSynced: lastBatchAddresses
-      }
+      yield { chronikTxs: remaining, addressesSynced: [] }
     }
   }
 
@@ -733,79 +713,85 @@ export class ChronikBlockchainClient {
     const perAddrCount = new Map<string, number>()
     addresses.forEach(a => perAddrCount.set(a.id, 0))
 
-    let txsToCreate: Prisma.TransactionUncheckedCreateInput[] = []
+    interface RowWithRaw { row: Prisma.TransactionUncheckedCreateInput, raw: Tx }
+    let toCommit: RowWithRaw[] = []
 
     try {
       const pfx = `${this.CHRONIK_MSG_PREFIX}[PARALLEL FETCHING]`
       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.chronikTxs.length} txs from chronik`)
-        // count per address before committing
-        const txRowsToCreate = await Promise.all(
-          batch.chronikTxs.map(({ tx, address }) =>
-            this.getTransactionFromChronikTransaction(tx, address)
-          )
+        if (batch.addressesSynced.length > 0) {
+          // marcador de slice => desmarca syncing
+          await setSyncingBatch(batch.addressesSynced, false)
+          continue
+        }
+
+        // são txs de fato
+        const pairsFromBatch: RowWithRaw[] = await Promise.all(
+          batch.chronikTxs.map(async ({ tx, address }) => {
+            const row = await this.getTransactionFromChronikTransaction(tx, address)
+            return { row, raw: tx }
+          })
         )
 
-        for (const tx of txRowsToCreate) {
-          perAddrCount.set(tx.addressId, (perAddrCount.get(tx.addressId) ?? 0) + 1)
+        for (const { row } of pairsFromBatch) {
+          perAddrCount.set(row.addressId, (perAddrCount.get(row.addressId) ?? 0) + 1)
         }
-        txsToCreate.push(...txRowsToCreate)
-        await setSyncingBatch(batch.addressesSynced, false)
 
-        if (txsToCreate.length >= DB_COMMIT_BATCH_SIZE) {
-          console.log(`${this.CHRONIK_MSG_PREFIX} committing batch to DB — txs=${txsToCreate.length}`)
-          const createdTxs = await createManyTransactions(txsToCreate)
+        toCommit.push(...pairsFromBatch)
+
+        if (toCommit.length >= DB_COMMIT_BATCH_SIZE) {
+          const commitPairs = toCommit.slice(0, DB_COMMIT_BATCH_SIZE)
+          toCommit = toCommit.slice(DB_COMMIT_BATCH_SIZE)
+
+          const rows = commitPairs.map(p => p.row)
+          const createdTxs = await createManyTransactions(rows)
           console.log(`${this.CHRONIK_MSG_PREFIX} committed — created=${createdTxs.length}`)
-          txsToCreate = []
 
-          // optional append for production addresses
           const createdForProd = createdTxs.filter(t => productionAddressesIds.includes(t.addressId))
           if (createdForProd.length > 0) {
             await appendTxsToFile(createdForProd as unknown as Prisma.TransactionCreateManyInput[])
           }
 
-          // broadcast/triggers after commit
           if (createdTxs.length > 0) {
+            const rawByHash = new Map(commitPairs.map(p => [p.raw.txid, p.raw]))
             const triggerBatch: BroadcastTxData[] = []
-            const rawByHash = new Map(
-              batch.chronikTxs.map(({ tx }) => [tx.txid, tx])
-            )
-            for (const createdTx of created) {
-              const raw = rawByHash.get(createdTx.txId) // ajuste: campo do seu modelo
-              if (!raw) continue
+            for (const createdTx of createdTxs) {
+              const raw = rawByHash.get(createdTx.hash)
+              if (raw == null) continue
               const bd = this.broadcastIncomingTx(createdTx.address.address, raw, createdTx)
               triggerBatch.push(bd)
             }
-            if (runTriggers) {
-              console.log(`${this.CHRONIK_MSG_PREFIX} executing trigger batch — broadcasts=${triggerBatch.length}`)
+            if (runTriggers && triggerBatch.length > 0) {
               await executeTriggersBatch(triggerBatch, this.networkId)
             }
           }
         }
       }
 
-      // final DB flush
-      if (txsToCreate.length > 0) {
-        console.log(`${this.CHRONIK_MSG_PREFIX} committing FINAL batch to DB — txs=${txsToCreate.length}`)
-        const createdTxs = await createManyTransactions(txsToCreate)
-        console.log(`${this.CHRONIK_MSG_PREFIX} committed FINAL — createdTxs=${createdTxs.length}`)
-        txsToCreate = []
+      // final DB flush (se sobrou menos que DB_COMMIT_BATCH_SIZE)
+      if (toCommit.length > 0) {
+        const rows = toCommit.map(p => p.row)
+        const createdTxs = await createManyTransactions(rows)
+        console.log(`${this.CHRONIK_MSG_PREFIX} committed FINAL — created=${createdTxs.length}`)
+        toCommit = []
 
         const createdForProd = createdTxs.filter(t => productionAddressesIds.includes(t.addressId))
         if (createdForProd.length > 0) {
           await appendTxsToFile(createdForProd as unknown as Prisma.TransactionCreateManyInput[])
         }
 
         if (createdTxs.length > 0) {
+          const rawByHash = new Map(toCommit.map(p => [p.raw.txid, p.raw]))
           const triggerBatch: BroadcastTxData[] = []
-          for (const tx of createdTxs) {
-            // WIP FIX
-            const bd = this.broadcastIncomingTx(tx.address.address, tx)
+          for (const createdTx of createdTxs) {
+            const raw = rawByHash.get(createdTx.hash)
+            if (raw == null) continue
+            const bd = this.broadcastIncomingTx(createdTx.address.address, raw, createdTx)
             triggerBatch.push(bd)
           }
-          if (runTriggers) {
-            console.log(`${this.CHRONIK_MSG_PREFIX} executing FINAL trigger batch — broadcasts=${triggerBatch.length}`)
+          if (runTriggers && triggerBatch.length > 0) {
             await executeTriggersBatch(triggerBatch, this.networkId)
           }
         }
@@ -830,9 +816,6 @@ export class ChronikBlockchainClient {
     const total = Object.values(successfulAddressesWithCount).reduce((p, c) => p + c, 0)
     console.log(`${this.CHRONIK_MSG_PREFIX} (PARALLEL) Finished syncing ${total} txs for ${addresses.length} addresses with ${failed.length} errors.`)
     console.timeEnd(`${this.CHRONIK_MSG_PREFIX} syncAddresses`)
-    if (failed.length > 0) {
-      console.log(`${this.CHRONIK_MSG_PREFIX} Failed addresses were:\n- ${Object.entries(failedAddressesWithErrors).map(([k, v]) => `${k}: ${v}`).join('\n- ')}`)
-    }
 
     return { failedAddressesWithErrors, successfulAddressesWithCount }
   }
