Issue 829 (#2453)

* feat: add income to account_books

Signed-off-by: Miles Zhang <mingchang555@hotmail.com>

* feat: calculate income when sync tx

Signed-off-by: Miles Zhang <mingchang555@hotmail.com>

* feat: use task to fill account_book data

Signed-off-by: Miles Zhang <mingchang555@hotmail.com>

---------

Signed-off-by: Miles Zhang <mingchang555@hotmail.com>

Author: zmcNotafraid

app/models/account_book.rb

@@ -11,9 +11,13 @@ class AccountBook < ApplicationRecord
 #  id                 :bigint           not null, primary key
 #  address_id         :bigint
 #  ckb_transaction_id :bigint
+#  income             :decimal(30, )
+#  block_number       :bigint
+#  tx_index           :integer
 #
 # Indexes
 #
 #  index_account_books_on_address_id_and_ckb_transaction_id  (address_id,ckb_transaction_id) UNIQUE
+#  index_account_books_on_block_number_and_tx_index          (block_number,tx_index)
 #  index_account_books_on_ckb_transaction_id                 (ckb_transaction_id)
 #

app/models/ckb_sync/new_node_data_processor.rb

@@ -8,7 +8,7 @@ class NewNodeDataProcessor
     include Redis::Objects
 
     value :reorg_started_at, global: true
-    attr_accessor :local_tip_block, :pending_raw_block, :ckb_txs, :target_block, :addrs_changes,
+    attr_accessor :local_tip_block, :pending_raw_block, :ckb_txs, :target_block, :target_block_number, :addrs_changes,
                   :outputs, :inputs, :outputs_data, :udt_address_ids, :contained_address_ids,
                   :contained_udt_ids, :cell_datas, :enable_cota, :token_transfer_ckb_tx_ids
 
@@ -23,7 +23,7 @@ def initialize(enable_cota = ENV["COTA_AGGREGATOR_URL"].present?)
     def call
       @local_tip_block = Block.recent.first
       tip_block_number = @tip_block_number = CkbSync::Api.instance.get_tip_block_number
-      target_block_number = local_tip_block.present? ? local_tip_block.number + 1 : 0
+      @target_block_number = local_tip_block.present? ? local_tip_block.number + 1 : 0
       puts "Offset #{@offset}" if @offset > 0
       return if target_block_number > tip_block_number - @offset.to_i
 
@@ -71,7 +71,7 @@ def process_block(node_block, refresh_balance: true)
 
         benchmark :process_ckb_txs, node_block, ckb_txs, contained_address_ids,
                   contained_udt_ids, tags, udt_address_ids
-        addrs_changes = Hash.new { |hash, key| hash[key] = {} }
+        @addrs_changes = Hash.new { |hash, key| hash[key] = {} }
 
         input_capacities, output_capacities = benchmark :build_cells_and_locks!, local_block, node_block, ckb_txs, inputs, outputs,
                                                         tags, udt_address_ids, contained_udt_ids, contained_address_ids, addrs_changes, token_transfer_ckb_tx_ids, cell_deps
@@ -678,8 +678,8 @@ def update_ckb_txs_rel_and_fee(
       ckb_txs.each do |tx|
         tx_id = tx["id"]
         full_tx_address_ids +=
-          contained_addr_ids[tx_index].to_a.map do |a|
-            { address_id: a, ckb_transaction_id: tx_id }
+          contained_addr_ids[tx_index].to_a.map do |address_id|
+            { address_id:, ckb_transaction_id: tx_id, income: addrs_changes[address_id][:balance_diff], block_number: target_block_number, tx_index: }
           end
         full_tx_udt_ids += contained_udt_ids[tx_index].to_a.map do |u|
           { udt_id: u, ckb_transaction_id: tx_id }

app/models/ckb_sync/transaction_parser.rb

@@ -2,7 +2,7 @@ module CkbSync
   class TransactionParser
     attr_reader :transaction, :extra_data
     attr_accessor :tx_attr, :cell_outputs_attrs, :cell_data_attrs, :cell_inputs_attrs, :cell_deps_attrs, :witnesses_attrs, :header_deps_attrs, :lock_script_attrs, :addresses_attrs,
-                  :account_books_attrs, :type_script_attrs
+                  :output_account_books_attrs, :type_script_attrs
 
     # {"transaction"=>{"version"=>"0x0", "cell_deps"=>[{"out_point"=>{"tx_hash"=>"0xcd52d714ddea04d2917892f16d47cbd0bbbb7d9ba281233ec4021f79fc34bccc", "index"=>"0x0"}, "dep_type"=>"code"}, {"out_point"=>{"tx_hash"=>"0x9154df4f7336402114d04495175b37390ce86a4906d2d4001cf02c3e6d97f39c", "index"=>"0x0"}, "dep_type"=>"code"}, {"out_point"=>{"tx_hash"=>"0xbcd73881ba53f1cd95d0c855395c4ffe6f54e041765d9ab7602d48a7cb71612e", "index"=>"0x0"}, "dep_type"=>"code"}, {"out_point"=>{"tx_hash"=>"0xf8de3bb47d055cdf460d93a2a6e1b05f7432f9777c8c474abf4eec1d4aee5d37", "index"=>"0x0"}, "dep_type"=>"dep_group"}, {"out_point"=>{"tx_hash"=>"0x053fdb4ed3181eab3a3a5f05693b53a8cdec0a24569e16369f444bac48be7de9", "index"=>"0x0"}, "dep_type"=>"code"}], "header_deps"=>[], "inputs"=>[{"since"=>"0x40000000669f4e30", "previous_output"=>{"tx_hash"=>"0x3d9a919a18d2cc2b64d2063626c75a6c97e87d2e8c30ec7bd33ef3ce14039934", "index"=>"0x0"}}, {"since"=>"0x0", "previous_output"=>{"tx_hash"=>"0x3d9a919a18d2cc2b64d2063626c75a6c97e87d2e8c30ec7bd33ef3ce14039934", "index"=>"0x1"}}, {"since"=>"0x0", "previous_output"=>{"tx_hash"=>"0x3d9a919a18d2cc2b64d2063626c75a6c97e87d2e8c30ec7bd33ef3ce14039934", "index"=>"0x2"}}], "outputs"=>[{"capacity"=>"0x7676d7e00", "lock"=>{"code_hash"=>"0x79f90bb5e892d80dd213439eeab551120eb417678824f282b4ffb5f21bad2e1e", "hash_type"=>"type", "args"=>"0x00c267a8b93cdae15fb06325f11a72b1047bd4d33c00"}, "type"=>{"code_hash"=>"0x1e44736436b406f8e48a30dfbddcf044feb0c9eebfe63b0f81cb5bb727d84854", "hash_type"=>"type", "args"=>"0x86c7429247beba7ddd6e4361bcdfc0510b0b644131e2afb7e486375249a01802"}}, {"capacity"=>"0x3691d6afc000", "lock"=>{"code_hash"=>"0x7f5a09b8bd0e85bcf2ccad96411ccba2f289748a1c16900b0635c2ed9126f288", "hash_type"=>"type", "args"=>"0x702359ea7f073558921eb50d8c1c77e92f760c8f8656bde4995f26b8963e2dd8f245705db4fe72be953e4f9ee3808a1700a578341aa80a8b2349c236c4af64e5e077710000000000"}, "type"=>nil}, {"capacity"=>"0xe529edc1ba", "lock"=>{"code_hash"=>"0x9bd7e06f3ecf4be0f2fcd2188b23f1b9fcc88e5d4b65a8637b17723bbda3cce8", "hash_type"=>"type", "args"=>"0xc267a8b93cdae15fb06325f11a72b1047bd4d33c"}, "type"=>nil}], "outputs_data"=>["0xa0cf6037bfc238b179b74a30a9b12e15a4fbdd8881aebc8e5a66a8b5b5c95f0a6d833f7d5c1f8130fa2688970b421a57fbdffeff55bae5b4676043c543355799abe60200de0991241ff42c0255e2e2f65d114951c0a144e89d35527c582adc6603ff1ea2e17771000000000000000000000000000000000000000000000000000000000000000000000000001f779faa1f6184b10c9d865f62bba000e5d54e00a5d4b98cd768e43e376f68421a6b49de900100007c777100000000000001", "0x", "0x"], "witnesses"=>["0xc1030000100000006900000069000000550000005500000010000000550000005500000041000000e74f7818a6d2d1dda76593b30973967cd15fc6853731ecc0a5cf42cdb81859005b4a22781fa67973ccfbd3c5d0c1ccdd62506e2e99a049c19adc972e3909511a0054030000000000005003000010000000480300004c030000380300001c0000006c01000070010000740100007801000034030000500100002c000000340000005400000074000000940000009c000000c0000000e4000000e80000000c010000e0777100000000001c0000000200000014000000715ab282b873b79a7be8b0e8c13c4e8966a52040f7cfb9cf096dc32d69cac2b6f884bb2b1a8bb01660f3edc613ccfbeb7f3506d6f245705db4fe72be953e4f9ee3808a1700a578341aa80a8b2349c236c4af64e51a6b49de900100006d833f7d5c1f8130fa2688970b421a57fbdffeff55bae5b4676043c543355799abe602006d833f7d5c1f8130fa2688970b421a57fbdffeff55bae5b4676043c543355799abe602000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000024ea893c4fa601a048b1d3a8de265fd8b442ba2a1ac37d85dfe320a7c8c2069a000000000000000004000000b80100004c4f05500d310a045d500ed295c313abef5ae2862c24393fdacc2e2e4c2ba76b43828b9d505cfc4cbb06c1083817aa14c6e06df9c300687a61fa0ec947fe79962fb9c557845058409c8c929c05fe19cc818cd0da6f0bf887cdb7ca4e465e85f7796ddd0e1528507119e4869f5a24e613de92605fb220451c0d1727d65f1c030815155c3cb7acd350a011fc6c115f153c24055880abd2a1253281e4e8a894e2a0546ea8799b057667507cee0dc2c4b8b171a0331b3f9ec02cf8906ef52a873523d2dbe3ff9e5f58699a4f01509d5c06f8f45bfaf59625237cb06c099a9c2a59752d5c1032e503ef39fbded2d7506989c3d61168d80ae0e27a2b2ca904f768cc698f783fd4f5d6c45c1b7dd66bbb5090d5dd9cd0b9f66197a69ea30e933982c98b99a067b37a8d0b7adc687d8c8db44f0150da2c4d3f8cc63b7b827c24c55098ec49cc88a9c81ff07abb0057f7d1dc05c9304f0350287fc2f5b5005ff558ae296f4e9cb354876829fa9562d215fbf7cd9536d5e0e550a03eb9492d55fc2d9d63faf8ea6121d66a86230f0752178ab5643a81abff8380508ad8ce2ac94cf885730b362ca4b81787bf85b48d7f72ef7816130bdc54433f644fe9040000000000000000000000", "0x10000000100000001000000010000000", "0x55000000100000005500000055000000410000001ccd614d2dbafb3384f6da8d50fc6cef21e0280c05397a93072d1560c81fd8bf347d299d7879e7b7682812fbb9fd5841c97c2f92b5c3e64737aa61c9fab7a01401"], "hash"=>"0x74758da9a59938724839e442d6e2b10b5c69e8fa398d4bd4d9b64fd311801ac1"}, "cycles"=>"0x4819ea", "size"=>"0x853", "fee"=>"0x853", "timestamp"=>"0x190de4b95b3"}
     def initialize(tx, extra_data = {})
@@ -14,9 +14,9 @@ def initialize(tx, extra_data = {})
       @cell_deps_attrs = []
       @witnesses_attrs = []
       @header_deps_attrs = []
+      @output_account_books_attrs = []
       @lock_script_attrs = Set.new
       @addresses_attrs = Set.new
-      @account_books_attrs = Set.new
       @type_script_attrs = Set.new
     end
 
@@ -69,7 +69,7 @@ def prepare_cell_outputs_params
         cell_type = CkbUtils.cell_type(output.type, output_data)
         @lock_script_attrs << output.lock.to_h.merge({ script_hash: output.lock.compute_hash })
         @addresses_attrs << { address_hash: CkbUtils.generate_address(output.lock), lock_hash: output.lock.compute_hash }
-        @account_books_attrs << { lock_script_hash: output.lock.compute_hash, tx_hash: transaction.hash }
+        @output_account_books_attrs << { lock_script_hash: output.lock.compute_hash, tx_hash: transaction.hash, capacity: output.capacity }
         @type_script_attrs << output.type.to_h.merge({ script_hash: output.type.compute_hash }) if output.type.present?
         @cell_outputs_attrs <<
           {

app/models/ckb_sync/transactions.rb

@@ -27,10 +27,11 @@ def import
         cell_deps_attrs = []
         witnesses_attrs = []
         header_deps_attrs = []
+        input_account_books_attrs = []
+        output_account_books_attrs = []
         lock_script_attrs = Set.new
         addresses_attrs = Set.new
         type_script_attrs = Set.new
-        account_books_attrs = Set.new
 
         group_parsers.each do |parser|
           tx_attrs << parser.tx_attr
@@ -40,10 +41,10 @@ def import
           cell_deps_attrs.concat(parser.cell_deps_attrs)
           header_deps_attrs.concat(parser.header_deps_attrs)
           cell_inputs_attrs.concat(parser.cell_inputs_attrs)
+          output_account_books_attrs.concat(parser.output_account_books_attrs)
           lock_script_attrs.merge(parser.lock_script_attrs)
           addresses_attrs.merge(parser.addresses_attrs)
           type_script_attrs.merge(parser.type_script_attrs)
-          account_books_attrs.merge(parser.account_books_attrs)
         end
         ApplicationRecord.transaction do
           tx_returnings = CkbTransaction.upsert_all(tx_attrs, unique_by: %i[tx_status tx_hash], returning: %i[id tx_hash])
@@ -61,11 +62,7 @@ def import
             type_script_returnings = TypeScript.upsert_all(type_script_attrs.to_a, unique_by: :script_hash, returning: %i[id script_hash])
             type_script_mappings = type_script_returnings.rows.to_h { |id, script_hash| [script_hash, id] }
           end
-          new_account_books_attrs =
-            account_books_attrs.to_a.map do |attr|
-              { ckb_transaction_id: tx_mappings[attr[:tx_hash]], address_id: address_mappings[attr[:lock_script_hash]] }
-            end
-          AccountBook.upsert_all(new_account_books_attrs, unique_by: %i[address_id ckb_transaction_id])
+
           new_cell_outputs_attrs =
             cell_outputs_attrs.map do |attr|
               attr.merge({ ckb_transaction_id: tx_mappings[attr[:tx_hash]], lock_script_id: lock_script_mappings[attr[:lock_script_hash]],
@@ -100,25 +97,45 @@ def import
           input_conditions = cell_inputs_attrs.filter do |input|
                                input[:previous_tx_hash] != CellOutput::SYSTEM_TX_HASH
                              end.map { |input| { tx_hash: input[:previous_tx_hash], cell_index: input[:previous_index] } }
-          input_returnings = batch_query_outputs(input_conditions, %i[id cell_type tx_hash cell_index capacity])
-          input_mappings = input_returnings.to_h { |id, cell_type, tx_hash, cell_index| ["#{tx_hash}-#{cell_index}", "#{cell_type}-#{id}"] }
+          input_returnings = batch_query_outputs(input_conditions, %i[id cell_type tx_hash cell_index address_id capacity])
+          input_mappings = input_returnings.to_h { |id, cell_type, tx_hash, cell_index, address_id, capacity| ["#{tx_hash}-#{cell_index}", { id:, cell_type:, address_id:, capacity: }] }
           new_cell_inputs_attrs =
             cell_inputs_attrs.map do |attr|
               attr[:ckb_transaction_id] = tx_mappings[attr[:tx_hash]]
               if attr[:previous_tx_hash] != CellOutput::SYSTEM_TX_HASH && input_mappings["#{attr[:previous_tx_hash]}-#{attr[:previous_index]}"].present?
-                cell_type, previous_cell_output_id = input_mappings["#{attr[:previous_tx_hash]}-#{attr[:previous_index]}"].split("-")
-                attr[:previous_cell_output_id] = previous_cell_output_id
-                attr[:cell_type] = cell_type
+                input_hash = input_mappings["#{attr[:previous_tx_hash]}-#{attr[:previous_index]}"]
+                attr[:previous_cell_output_id] = input_hash[:id]
+                attr[:cell_type] = input_hash[:cell_type]
+                input_account_books_attrs << { ckb_transaction_id: attr[:ckb_transaction_id], address_id: input_hash[:address_id], capacity: input_hash[:capacity] }
               end
               attr.except(:tx_hash)
             end
           CellInput.upsert_all(new_cell_inputs_attrs, unique_by: %i[ckb_transaction_id index])
+          new_output_account_books_attrs =
+            output_account_books_attrs.map do |attr|
+              { ckb_transaction_id: tx_mappings[attr[:tx_hash]], address_id: address_mappings[attr[:lock_script_hash]], capacity: attr[:capacity] }
+            end
+          account_books_attrs = calculate_income(new_output_account_books_attrs, input_account_books_attrs)
+          AccountBook.upsert_all(account_books_attrs, unique_by: %i[address_id ckb_transaction_id])
         end
       end
     end
 
     private
 
+    def calculate_income(outputs, inputs)
+      grouped_outputs = outputs.group_by { |o| [o[:ckb_transaction_id], o[:address_id]] }
+      grouped_inputs = inputs.group_by { |i| [i[:ckb_transaction_id], i[:address_id]] }
+
+      all_keys = (grouped_outputs.keys + grouped_inputs.keys).uniq
+
+      all_keys.map do |key|
+        out_cap = grouped_outputs[key]&.sum { |o| o[:capacity] } || 0
+        in_cap = grouped_inputs[key]&.sum { |i| i[:capacity] } || 0
+        { address_id: key[1], ckb_transaction_id: key[0], income: out_cap - in_cap }
+      end
+    end
+
     def batch_query_outputs(conditions, returnings = %i[id cell_type tx_hash cell_index])
       relation = CellOutput.none
 

db/migrate/20250318021630_add_income_to_account_book.rb

@@ -0,0 +1,10 @@
+class AddIncomeToAccountBook < ActiveRecord::Migration[7.0]
+  def change
+     execute("SET statement_timeout = 0;")
+     add_column :account_books, :income, :decimal, precision: 30
+     add_column :account_books, :block_number, :bigint
+     add_column :account_books, :tx_index, :integer
+
+     add_index :account_books, [:block_number, :tx_index]
+  end
+end

db/structure.sql

@@ -324,7 +324,10 @@ SET default_table_access_method = heap;
 CREATE TABLE public.account_books (
     id bigint NOT NULL,
     address_id bigint,
-    ckb_transaction_id bigint
+    ckb_transaction_id bigint,
+    income numeric(30,0),
+    block_number bigint,
+    tx_index integer
 );
 
 
@@ -4719,6 +4722,13 @@ CREATE INDEX idx_cell_inputs_on_previous_tx_hash_and_previous_index ON public.ce
 CREATE UNIQUE INDEX index_account_books_on_address_id_and_ckb_transaction_id ON public.account_books USING btree (address_id, ckb_transaction_id);
 
 
+--
+-- Name: index_account_books_on_block_number_and_tx_index; Type: INDEX; Schema: public; Owner: -
+--
+
+CREATE INDEX index_account_books_on_block_number_and_tx_index ON public.account_books USING btree (block_number, tx_index);
+
+
 --
 -- Name: index_account_books_on_ckb_transaction_id; Type: INDEX; Schema: public; Owner: -
 --
@@ -6389,6 +6399,7 @@ INSERT INTO "schema_migrations" (version) VALUES
 ('20250108053433'),
 ('20250126022459'),
 ('20250218062041'),
-('20250311084903');
+('20250311084903'),
+('20250318021630');
 
 

lib/tasks/migration/fill_account_book_block_number_and_tx_index.rake

@@ -0,0 +1,38 @@
+namespace :migration do
+  desc "Usage: RAILS_ENV=production bundle exec rake migration:fill_account_book_block_number_and_tx_index[0,1000000]"
+  task :fill_account_book_block_number_and_tx_index, %i[start_block end_block] => :environment do |_, args|
+    $missed_tx_ids = []
+    (args[:start_block].to_i..args[:end_block].to_i).to_a.each do |block_number|
+      attrs = Set.new
+      CkbTransaction.includes(:inputs, :outputs).where(block_number:).where(is_cellbase: false).each do |tx|
+        outputs = tx.outputs.pluck(:address_id, :capacity).group_by { |item| item[0] }.
+          transform_values { |values| values.sum { |v| v[1] } }
+        inputs = tx.inputs.pluck(:address_id, :capacity).group_by { |item| item[0] }.
+          transform_values { |values| values.sum { |v| v[1] } }
+        address_ids = (outputs.keys + inputs.keys).uniq
+        exists = ensure_all_data_exists(address_ids, tx.id)
+        if exists
+          address_ids.each do |address_id|
+            income = (outputs[address_id] || 0) - (inputs[address_id] || 0)
+            attrs << { address_id:, ckb_transaction_id: tx.id, income:, block_number: tx.block_number, tx_index: tx.tx_index }
+          end
+        else
+          $missed_tx_ids << tx.id
+        end
+      end
+      AccountBook.upsert_all(attrs.to_a, unique_by: %i[address_id ckb_transaction_id]) if attrs.present?
+    end
+
+    puts $missed_tx_ids.join(",")
+    puts "done"
+  end
+  def ensure_all_data_exists(address_ids, tx_id)
+    data =
+      address_ids.map do |address_id|
+        { address_id:, ckb_transaction_id: tx_id }
+      end
+    query_conditions = data.map { |d| "(address_id = #{d[:address_id]} AND ckb_transaction_id = #{d[:ckb_transaction_id]})" }.join(" OR ")
+    existing_records = AccountBook.where(query_conditions).pluck(:address_id, :ckb_transaction_id)
+    data.map(&:values).to_set.subset?(existing_records.to_set)
+  end
+end