message: [indexer]add delete pending txn handle and fix event id problem
diff: @@ -19,6 +19,7 @@ pub struct IndexConfig {
  pub block_index: String,
  pub uncle_block_index: String,
  pub txn_info_index: String,
  + pub pending_txn_index: String,
  pub txn_event_index: String,
  }
  @@ -28,6 +29,7 @@ impl IndexConfig {
  block_index: format!("{}.blocks", prefix.as_ref()),
  uncle_block_index: format!("{}.uncle_blocks", prefix.as_ref()),
  txn_info_index: format!("{}.txn_infos", prefix.as_ref()),
  + pending_txn_index: format!("{}.pending_txns", prefix.as_ref()),
  txn_event_index: format!("{}.txn_events", prefix.as_ref()),
  }
  }
  @@ -39,6 +41,7 @@ impl Default for IndexConfig {
  block_index: "blocks".to_string(),
  uncle_block_index: "uncle_blocks".to_string(),
  txn_info_index: "txn_infos".to_string(),
  + pending_txn_index: "pending_txns".to_string(),
  txn_event_index: "txn_events".to_string(),
  }
  }
  @@ -96,10 +99,12 @@ impl EsSinker {
  let uncle_block_index = self.config.uncle_block_index.as_str();
  let txn_info_index = self.config.txn_info_index.as_str();
  let txn_event_index = self.config.txn_event_index.as_str();
  + let pending_txn_index = self.config.pending_txn_index.as_str();
  self.create_index_if_not_exists(block_index).await?;
  self.create_index_if_not_exists(uncle_block_index).await?;
  self.create_index_if_not_exists(txn_info_index).await?;
  self.create_index_if_not_exists(txn_event_index).await?;
  + self.create_index_if_not_exists(pending_txn_index).await?;
  let tip = self.get_remote_tip_header().await?;
  self.state.write().await.tip = tip.clone();
  if let Some(tip_info) = tip {
  @@ -272,6 +277,7 @@ impl EsSinker {
  let mut bulk_operations = BulkOperations::new();
  let block_index = self.config.block_index.as_str();
  let txn_info_index = self.config.txn_info_index.as_str();
  + let pending_txn_index = self.config.pending_txn_index.as_str();
  let uncle_index = self.config.uncle_block_index.as_str();
  let event_index = self.config.txn_event_index.as_str();
  @@ -291,18 +297,17 @@ impl EsSinker {
  .id(txn_data.info.transaction_hash.to_string())
  .index(txn_info_index),
  )?;
  + bulk_operations.push(
  + BulkOperation::<()>::delete(txn_data.info.transaction_hash.to_string())
  + .index(pending_txn_index),
  + )?;
  if !txn_data.events.is_empty() {
  // event_vec.extend_from_slice(txn_data.events.as_slice());
  for event in txn_data.events {
  let mut event_data = EventData::from(event.clone());
  event_data.timestamp = txn_data.timestamp;
  - if let Some(tag_name) = event_data.clone().tag_name {
  - let id = format!("{}_{}", tag_name, event.event_seq_number);
  bulk_operations
  - .push(BulkOperation::index(event_data).id(id).index(event_index))?;
  - } else {
  - warn!("other event: {}", event_data.type_tag);
  - }
  + .push(BulkOperation::index(event_data).index(event_index))?;
  }
  }
  }
  