Store transaction inputs/outputs address as string (#1116)

* Store transaction inputs/outputs address as string

This release some constraints on the database and improve the performances.

* Reduce concurrency

For the chronik calls, and the db batch sizes. This avoids too long db transactions from impacting the performances and allow to start working faster on the chronik transactions.

* Update the logic in createManyTransactions

This avoid initiating a DB transaction for each tx, and favor bulk operations instead. Since it also removes the batch processing this removes the connection limits to the db issue entirely when an address has too many txs.

* Optimize getTransactionFromChronikTransaction

This is a 2-step optimization:
 - remove unneeded async in satoshisToUnit() callsites, this reduce locks
 - optimize the inputs and outputs processing by avoiding looping several times, avoid copying arrays and doing redundant formatting/checks.

This also fixes a logging error and a loop exit condition that could potentially cause an infinite loop. Inspired by 6298e84 and c4cb339.

This is a ~500x improvement on my machine.

* fix: one network at a time

* fix: current prices after initial sync

* Fix the chronik service tests

* Remove TransactionOutput from the database

It's unclear what the use case is and it was added in #1110 to avoid having to do it later.
It appears that this table is costly as it contains a lot of outputs.

Let's remove it for now to favor disk size and performance. We can add it back later if we need it.

---------

Co-authored-by: Estevão <estevao@chedieck.com>

Author: Fabcien

constants/index.ts

@@ -277,9 +277,9 @@ 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 = 128
-export const TX_EMIT_BATCH_SIZE = 2_000 // for our generator, not chronik
-export const DB_COMMIT_BATCH_SIZE = 2_000 // tamanho dos lotes para commit no DB
+export const INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY = 16
+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 TRIGGER_POST_CONCURRENCY = 100
 export const TRIGGER_EMAIL_CONCURRENCY = 100

jobs/initJobs.ts

@@ -16,19 +16,7 @@ const main = async (): Promise<void> => {
   await blockchainQueue.obliterate({ force: true })
   await cleanupQueue.obliterate({ force: true })
 
-  await pricesQueue.add('syncCurrentPrices',
-    {},
-    {
-      jobId: 'syncCurrentPrices',
-      removeOnFail: false,
-      repeat: {
-        every: CURRENT_PRICE_REPEAT_DELAY
-      }
-    }
-  )
-
-  await syncCurrentPricesWorker(pricesQueue.name)
-
+  // Start blockchain sync first; current prices job starts only after it completes
   await blockchainQueue.add('syncBlockchainAndPrices',
     {},
     {
@@ -37,7 +25,20 @@ const main = async (): Promise<void> => {
       removeOnFail: true
     }
   )
-  await syncBlockchainAndPricesWorker(blockchainQueue.name)
+  void await syncBlockchainAndPricesWorker(blockchainQueue.name, async () => {
+    await pricesQueue.add('syncCurrentPrices',
+      {},
+      {
+        jobId: 'syncCurrentPrices',
+        removeOnFail: false,
+        repeat: {
+          every: CURRENT_PRICE_REPEAT_DELAY
+        }
+      }
+    )
+    await syncCurrentPricesWorker(pricesQueue.name)
+    console.log('Current prices sync job started after blockchain sync completion.')
+  })
 
   await cleanupQueue.add(
     'cleanupClientPayments',

jobs/workers.ts

@@ -31,7 +31,7 @@ export const syncCurrentPricesWorker = async (queueName: string): Promise<void>
   })
 }
 
-export const syncBlockchainAndPricesWorker = async (queueName: string): Promise<void> => {
+export const syncBlockchainAndPricesWorker = async (queueName: string, onComplete?: () => Promise<void> | void): Promise<void> => {
   const worker = new Worker(
     queueName,
     async (job) => {
@@ -53,6 +53,7 @@ export const syncBlockchainAndPricesWorker = async (queueName: string): Promise<
       await multiBlockchainClient.destroy()
       console.log('Done.')
       console.log(`job ${job.id as string}: blockchain + prices sync finished`)
+      await onComplete?.()
     })()
   })
 

pages/api/address/balance/[address].ts

@@ -17,7 +17,7 @@ export default async (req: NextApiRequest, res: NextApiResponse): Promise<void>
     try {
       const address = parseAddress(req.query.address as string)
       const response = await multiBlockchainClient.getBalance(address)
-      const balance = await satoshisToUnit(response, xecaddr.detectAddressFormat(address))
+      const balance = satoshisToUnit(response, xecaddr.detectAddressFormat(address))
       res.status(200).send(balance)
     } catch (err: any) {
       switch (err.message) {

prisma-local/migrations/20260223150000_change_transaction_input_output_address_to_string/migration.sql

@@ -0,0 +1,44 @@
+-- Drop foreign key constraints
+ALTER TABLE `TransactionInput` DROP FOREIGN KEY `TransactionInput_addressId_fkey`;
+ALTER TABLE `TransactionOutput` DROP FOREIGN KEY `TransactionOutput_addressId_fkey`;
+
+-- Drop old indexes
+ALTER TABLE `TransactionInput` DROP INDEX `TransactionInput_addressId_idx`;
+ALTER TABLE `TransactionOutput` DROP INDEX `TransactionOutput_addressId_idx`;
+
+-- Add new address column (nullable temporarily)
+ALTER TABLE `TransactionInput` ADD COLUMN `address` VARCHAR(255) NULL;
+ALTER TABLE `TransactionOutput` ADD COLUMN `address` VARCHAR(255) NULL;
+
+-- Populate address column from Address table
+UPDATE `TransactionInput` ti
+INNER JOIN `Address` a ON ti.`addressId` = a.`id`
+SET ti.`address` = a.`address`;
+
+UPDATE `TransactionOutput` tout
+INNER JOIN `Address` a ON tout.`addressId` = a.`id`
+SET tout.`address` = a.`address`;
+
+-- Make address column NOT NULL
+ALTER TABLE `TransactionInput` MODIFY COLUMN `address` VARCHAR(255) NOT NULL;
+ALTER TABLE `TransactionOutput` MODIFY COLUMN `address` VARCHAR(255) NOT NULL;
+
+-- Delete Address entries that were only used by TransactionInput/TransactionOutput
+-- These addresses are no longer needed since we store addresses as strings
+DELETE a FROM `Address` a
+WHERE (
+  -- Address is referenced in TransactionInput or TransactionOutput
+  EXISTS (SELECT 1 FROM `TransactionInput` ti WHERE ti.`addressId` = a.`id`)
+  OR EXISTS (SELECT 1 FROM `TransactionOutput` tout WHERE tout.`addressId` = a.`id`)
+)
+AND NOT (
+  -- But exclude addresses that are still used elsewhere
+  EXISTS (SELECT 1 FROM `Transaction` t WHERE t.`addressId` = a.`id`)
+  OR EXISTS (SELECT 1 FROM `AddressesOnUserProfiles` aup WHERE aup.`addressId` = a.`id`)
+  OR EXISTS (SELECT 1 FROM `AddressesOnButtons` ab WHERE ab.`addressId` = a.`id`)
+  OR EXISTS (SELECT 1 FROM `ClientPayment` cp WHERE cp.`addressString` = a.`address`)
+);
+
+-- Drop old addressId column
+ALTER TABLE `TransactionInput` DROP COLUMN `addressId`;
+ALTER TABLE `TransactionOutput` DROP COLUMN `addressId`;

prisma-local/migrations/20260227163713_remove_transaction_output/migration.sql

@@ -0,0 +1,4 @@
+-- Remove TransactionOutput table
+-- This table is being removed as its use case is unclear and it impacts performance
+
+DROP TABLE IF EXISTS `TransactionOutput`;

prisma-local/schema.prisma

@@ -23,8 +23,6 @@ model Address {
   paybuttons         AddressesOnButtons[]
   transactions       Transaction[]
   clientPayments     ClientPayment[]
-  transactionInputs  TransactionInput[]
-  transactionOutputs TransactionOutput[]
 
   @@index([networkId], map: "Address_networkId_fkey")
 }
@@ -81,7 +79,6 @@ model Transaction {
   prices    PricesOnTransactions[]
   invoices  Invoice[]
   inputs    TransactionInput[]
-  outputs   TransactionOutput[]
 
   createdAt DateTime                  @default(now())
   updatedAt DateTime                  @updatedAt
@@ -93,28 +90,14 @@ model Transaction {
 model TransactionInput {
   id            String      @id @default(dbgenerated("(uuid())"))
   transactionId String
-  addressId     String
+  address       String      @db.VarChar(255)
   index         Int
   transaction   Transaction @relation(fields: [transactionId], references: [id], onDelete: Cascade)
-  address       Address     @relation(fields: [addressId], references: [id], onDelete: Cascade)
   amount        Decimal     @db.Decimal(24, 8)
 
   @@index([transactionId])
-  @@index([addressId])
 }
 
-model TransactionOutput {
-  id            String      @id @default(dbgenerated("(uuid())"))
-  transactionId String
-  addressId     String
-  index         Int
-  transaction   Transaction @relation(fields: [transactionId], references: [id], onDelete: Cascade)
-  address       Address     @relation(fields: [addressId], references: [id], onDelete: Cascade)
-  amount        Decimal     @db.Decimal(24, 8)
-
-  @@index([transactionId])
-  @@index([addressId])
-}
 
 model Wallet {
   id             String                    @id @default(dbgenerated("(uuid())"))