Merge pull request #2496 from nervosnetwork/testnet

Deploy to mainnet

Author: zmcNotafraid

.rubocop.yml

@@ -6,11 +6,7 @@ def index
           expires_in 15.minutes, public: true, stale_while_revalidate: 5.minutes, stale_if_error: 5.minutes
           statistics = FiberStatistic.order(created_at_unixtimestamp: :desc).limit(7)
 
-          render json: {
-            data: statistics.map do |statistic|
-              statistic.attributes.except("id", "created_at", "updated_at").transform_values(&:to_s)
-            end,
-          }
+          render json: { data: statistics.map(&:as_json) }
         end
 
         def show

app/controllers/api/v2/fiber/statistics_controller.rb

@@ -10,7 +10,7 @@ class NewNodeDataProcessor
     value :reorg_started_at, global: true
     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
+                  :contained_udt_ids, :cell_datas, :enable_cota, :token_transfer_ckb_tx_ids, :addr_tx_changes
 
     def initialize(enable_cota = ENV["COTA_AGGREGATOR_URL"].present?)
       @enable_cota = enable_cota
@@ -72,6 +72,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] = {} }
+        @addr_tx_changes = Hash.new { |h, k| h[k] = Hash.new(0) }
 
         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
@@ -526,7 +527,7 @@ def update_addresses_info(addrs_change, local_block, refresh_balance)
             ckb_transactions_count: addr.ckb_transactions_count,
             live_cells_count: addr.live_cells_count,
             dao_transactions_count: addr.dao_transactions_count,
-            last_updated_block_number: local_block.number
+            last_updated_block_number: local_block.number,
           },
           address_id: addr.id,
           block_id: local_block.id,
@@ -688,7 +689,7 @@ def update_ckb_txs_rel_and_fee(
         tx_id = tx["id"]
         full_tx_address_ids +=
           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: }
+            { address_id:, ckb_transaction_id: tx_id, income: addr_tx_changes[tx_index][address_id], 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 }
@@ -931,8 +932,8 @@ def build_cell_inputs(
             type_hash = attributes[:type_hash]
             data = attributes[:data]
             change_rec = addrs_changes[address_id]
-            # change_rec.with_defaults!
 
+            addr_tx_changes[tx_index][address_id] -= capacity
             change_rec[:balance_diff] ||= 0
             change_rec[:balance_diff]  -= capacity
             change_rec[:balance_occupied_diff] ||= 0
@@ -1005,6 +1006,7 @@ def build_cell_outputs!(
           address_id = address.id
           cell_data = node_block.transactions[tx_index].outputs_data[cell_index]
           change_rec = addrs_changes[address_id]
+          addr_tx_changes[tx_index][address_id] += item.capacity
 
           change_rec[:balance_diff] ||= 0
           change_rec[:balance_diff] += item.capacity

app/models/ckb_sync/new_node_data_processor.rb

@@ -76,9 +76,12 @@ def deleted_at_timestamp
 #  fee_rate_of_node1               :decimal(30, )    default(0)
 #  fee_rate_of_node2               :decimal(30, )    default(0)
 #  deleted_at                      :datetime
+#  cell_output_id                  :bigint
+#  address_id                      :bigint
 #
 # Indexes
 #
+#  index_fiber_graph_channels_on_address_id        (address_id)
 #  index_fiber_graph_channels_on_channel_outpoint  (channel_outpoint) UNIQUE
 #  index_fiber_graph_channels_on_deleted_at        (deleted_at)
 #

app/models/fiber_graph_channel.rb

@@ -1,11 +1,100 @@
 class FiberStatistic < ApplicationRecord
-  VALID_INDICATORS = %w(total_nodes total_channels total_liquidity created_at_unixtimestamp).freeze
+  include AttrLogics
+
+  VALID_INDICATORS = %w(total_nodes total_channels total_capacity created_at_unixtimestamp).freeze
 
   scope :filter_by_indicator, ->(indicator) {
     raise ArgumentError, "Invalid indicator" unless VALID_INDICATORS.include?(indicator.to_s)
 
     select(indicator, :created_at_unixtimestamp)
   }
+
+  define_logic(:total_nodes) { FiberGraphNode.count }
+  define_logic(:total_channels) { FiberGraphChannel.count }
+  define_logic(:total_capacity) { FiberGraphChannel.sum(:capacity) }
+
+  define_logic :total_liquidity do
+    result = Hash.new { |h, k| h[k] = 0.0 }
+
+    FiberGraphNode.find_each do |node|
+      channels = FiberGraphChannel.with_deleted.where(node1: node.node_id).or(
+        FiberGraphChannel.with_deleted.where(node2: node.node_id),
+      ).where(closed_transaction_id: nil)
+
+      channels.each do |channel|
+        funding_cell = channel.funding_cell
+        if funding_cell.cell_type.in?(%w(udt xudt xudt_compatible))
+          result[funding_cell.type_hash] += funding_cell.udt_amount
+        else
+          result[""] += channel.capacity
+        end
+      end
+    end
+
+    CkbUtils.hash_value_to_s(result)
+  end
+
+  define_logic :mean_value_locked do
+    total_channels.zero? ? 0.0 : total_capacity.to_f / total_channels
+  end
+
+  define_logic :mean_fee_rate do
+    FiberGraphChannel.average("fee_rate_of_node1 + fee_rate_of_node2") || 0.0
+  end
+
+  define_logic :medium_value_locked do
+    capacities = FiberGraphChannel.pluck(:capacity).compact
+    calculate_median(capacities)
+  end
+
+  define_logic :medium_fee_rate do
+    fee_rate_of_node1 = FiberGraphChannel.pluck(:fee_rate_of_node1).compact
+    fee_rate_of_node2 = FiberGraphChannel.pluck(:fee_rate_of_node2).compact
+    combined_fee_rates = fee_rate_of_node1 + fee_rate_of_node2
+    calculate_median(combined_fee_rates)
+  end
+
+  def calculate_median(array)
+    sorted = array.sort
+    count = sorted.size
+    return nil if count.zero?
+
+    if count.odd?
+      sorted[count / 2]
+    else
+      (sorted[(count / 2) - 1] + sorted[count / 2]).to_f / 2
+    end
+  end
+
+  def parsed_total_liquidity
+    total_liquidity&.map do |type_hash, amount|
+      if type_hash.present?
+        udt_info = Udt.find_by(type_hash: type_hash)
+        CkbUtils.hash_value_to_s(
+          symbol: udt_info.symbol,
+          amount: amount,
+          decimal: udt_info.decimal,
+          type_hash: type_hash,
+          published: !!udt_info.published,
+        )
+      else
+        CkbUtils.hash_value_to_s(symbol: "CKB", amount: amount)
+      end
+    end
+  end
+
+  def as_json(_options = {})
+    CkbUtils.hash_value_to_s(
+      total_nodes:,
+      total_channels:,
+      total_capacity:,
+      mean_value_locked:,
+      mean_fee_rate:,
+      medium_value_locked:,
+      medium_fee_rate:,
+      created_at_unixtimestamp:,
+    ).merge(total_liquidity: parsed_total_liquidity)
+  end
 end
 
 # == Schema Information
@@ -15,14 +104,15 @@ class FiberStatistic < ApplicationRecord
 #  id                       :bigint           not null, primary key
 #  total_nodes              :integer
 #  total_channels           :integer
-#  total_liquidity          :bigint
+#  total_capacity           :bigint
 #  mean_value_locked        :bigint
 #  mean_fee_rate            :integer
 #  medium_value_locked      :bigint
 #  medium_fee_rate          :integer
 #  created_at_unixtimestamp :integer
 #  created_at               :datetime         not null
 #  updated_at               :datetime         not null
+#  total_liquidity          :jsonb
 #
 # Indexes
 #

app/models/fiber_statistic.rb

@@ -122,41 +122,9 @@ def extract_peer_id(addresses)
   end
 
   def compute_statistic
-    total_nodes = FiberGraphNode.count
-    total_channels = FiberGraphChannel.count
-    # 资金总量
-    total_liquidity = FiberGraphChannel.sum(:capacity)
-    # 资金均值
-    mean_value_locked = total_channels.zero? ? 0.0 : total_liquidity.to_f / total_channels
-    # fee 均值
-    mean_fee_rate = FiberGraphChannel.average("fee_rate_of_node1 + fee_rate_of_node2") || 0.0
-    # 获取 capacity 的数据
-    capacities = FiberGraphChannel.pluck(:capacity).compact
-    # 获取 fee_rate_of_node1 和 fee_rate_of_node2 的数据并合并
-    fee_rate_of_node1 = FiberGraphChannel.pluck(:fee_rate_of_node1).compact
-    fee_rate_of_node2 = FiberGraphChannel.pluck(:fee_rate_of_node2).compact
-    combined_fee_rates = fee_rate_of_node1 + fee_rate_of_node2
-    # 计算中位数
-    medium_value_locked = calculate_median(capacities)
-    medium_fee_rate = calculate_median(combined_fee_rates)
     created_at_unixtimestamp = Time.now.beginning_of_day.to_i
-    FiberStatistic.upsert(
-      { total_nodes:, total_channels:, total_liquidity:,
-        mean_value_locked:, mean_fee_rate:, medium_value_locked:,
-        medium_fee_rate:, created_at_unixtimestamp: }, unique_by: %i[created_at_unixtimestamp]
-    )
-  end
-
-  def calculate_median(array)
-    sorted = array.sort
-    count = sorted.size
-    return nil if count.zero?
-
-    if count.odd?
-      sorted[count / 2] # 奇数个，取中间值
-    else
-      (sorted[(count / 2) - 1] + sorted[count / 2]).to_f / 2 # 偶数个，取中间两个的平均值
-    end
+    statistic = FiberStatistic.find_or_create_by!(created_at_unixtimestamp:)
+    statistic.reset_all!
   end
 
   def rpc

app/workers/fiber_graph_detect_worker.rb

@@ -0,0 +1,6 @@
+class AddTotalLiquidityToFiberStatistics < ActiveRecord::Migration[7.0]
+  def change
+    rename_column :fiber_statistics, :total_liquidity, :total_capacity
+    add_column :fiber_statistics, :total_liquidity, :jsonb
+  end
+end

db/migrate/20250408020030_add_total_liquidity_to_fiber_statistics.rb

@@ -1770,7 +1770,9 @@ CREATE TABLE public.fiber_graph_channels (
     last_updated_timestamp_of_node2 bigint,
     fee_rate_of_node1 numeric(30,0) DEFAULT 0.0,
     fee_rate_of_node2 numeric(30,0) DEFAULT 0.0,
-    deleted_at timestamp(6) without time zone
+    deleted_at timestamp(6) without time zone,
+    cell_output_id bigint,
+    address_id bigint
 );
 
 
@@ -1876,14 +1878,15 @@ CREATE TABLE public.fiber_statistics (
     id bigint NOT NULL,
     total_nodes integer,
     total_channels integer,
-    total_liquidity bigint,
+    total_capacity bigint,
     mean_value_locked bigint,
     mean_fee_rate integer,
     medium_value_locked bigint,
     medium_fee_rate integer,
     created_at_unixtimestamp integer,
     created_at timestamp(6) without time zone NOT NULL,
-    updated_at timestamp(6) without time zone NOT NULL
+    updated_at timestamp(6) without time zone NOT NULL,
+    total_liquidity jsonb
 );
 
 
@@ -5114,6 +5117,13 @@ CREATE INDEX index_fiber_channels_on_fiber_peer_id ON public.fiber_channels USIN
 CREATE UNIQUE INDEX index_fiber_channels_on_peer_id_and_channel_id ON public.fiber_channels USING btree (peer_id, channel_id);
 
 
+--
+-- Name: index_fiber_graph_channels_on_address_id; Type: INDEX; Schema: public; Owner: -
+--
+
+CREATE INDEX index_fiber_graph_channels_on_address_id ON public.fiber_graph_channels USING btree (address_id);
+
+
 --
 -- Name: index_fiber_graph_channels_on_channel_outpoint; Type: INDEX; Schema: public; Owner: -
 --
@@ -6401,6 +6411,8 @@ INSERT INTO "schema_migrations" (version) VALUES
 ('20250218062041'),
 ('20250311084903'),
 ('20250318021630'),
-('20250403090946');
+('20250402032340'),
+('20250403090946'),
+('20250408020030');
 
 

db/structure.sql

@@ -0,0 +1,28 @@
+namespace :migration do
+  desc "Usage: RAILS_ENV=production bundle exec rake migration:fix_account_book_income[0,1000000]"
+  task :fix_account_book_income, %i[start_block end_block] => :environment do |_, args|
+    (args[:start_block].to_i..args[:end_block].to_i).to_a.each_slice(1000).each do |range|
+      block_numbers = AccountBook.where("block_number >= ? and block_number <= ?", range[0], range[-1]).
+        select(:block_number, :address_id).
+        group(:block_number, :address_id).
+        having("COUNT(*) > 1").
+        distinct.
+        pluck(:block_number)
+      attrs = Set.new
+      CkbTransaction.joins(:block).includes(:inputs, :outputs).where(block: { number: block_numbers }).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
+        address_ids.each do |address_id|
+          income = (outputs[address_id] || 0) - (inputs[address_id] || 0)
+          attrs << { address_id:, ckb_transaction_id: tx.id, income: }
+        end
+      end
+      AccountBook.upsert_all(attrs.to_a, unique_by: %i[address_id ckb_transaction_id]) if attrs.present?
+    end
+
+    puts "done"
+  end
+end