[Indexer] Migrate to grpc processors (aptos-labs#8841)

* add from old indexer

* fix end version

* tweak comments on crash

* strip out coin type in a different way

* fix side effect cause by rust upgrade

* cargo lock update

Author: bowenyang007

Cargo.lock

@@ -68,13 +68,12 @@ impl CurrentDelegatorBalance {
         if let Some(pool_balance) = active_pool_to_staking_pool.get(&table_handle) {
             let pool_address = pool_balance.staking_pool_address.clone();
             let delegator_address = standardize_address(&write_table_item.key.to_string());
-            let data: &aptos_api_types::transaction::DecodedTableData =
-                write_table_item.data.as_ref().unwrap_or_else(|| {
-                    panic!(
+            let data = write_table_item.data.as_ref().unwrap_or_else(|| {
+                panic!(
                     "This table item should be an active share item, table_item {:?}, version {}",
                     write_table_item, txn_version
                 )
-                });
+            });
             let shares = data
                 .value
                 .as_str()
@@ -130,13 +129,12 @@ impl CurrentDelegatorBalance {
                 },
             };
             let delegator_address = standardize_address(&write_table_item.key.to_string());
-            let data: &aptos_api_types::transaction::DecodedTableData =
-                write_table_item.data.as_ref().unwrap_or_else(|| {
-                    panic!(
-                        "This table item should be an active share item, table_item {:?}, version {}",
-                        write_table_item, txn_version
-                    )
-                });
+            let data = write_table_item.data.as_ref().unwrap_or_else(|| {
+                panic!(
+                    "This table item should be an active share item, table_item {:?}, version {}",
+                    write_table_item, txn_version
+                )
+            });
             let shares = data
                 .value
                 .as_str()

crates/indexer/src/models/stake_models/delegator_balances.rs

@@ -126,8 +126,7 @@ impl DelegatorPool {
         if let Some(StakeResource::DelegationPool(inner)) =
             StakeResource::from_write_resource(write_resource, txn_version)?
         {
-            let staking_pool_address: String =
-                standardize_address(&write_resource.address.to_string());
+            let staking_pool_address = standardize_address(&write_resource.address.to_string());
             let total_coins = inner.active_shares.total_coins;
             let total_shares =
                 &inner.active_shares.total_shares / &inner.active_shares.scaling_factor;

crates/indexer/src/models/stake_models/delegator_pools.rs

@@ -43,7 +43,14 @@ pub fn ensure_not_negative(val: BigDecimal) -> BigDecimal {
 }
 
 pub fn parse_timestamp(ts: u64, version: i64) -> chrono::NaiveDateTime {
-    chrono::NaiveDateTime::from_timestamp_opt((ts / 1000000) as i64, 0)
+    let seconds = ts / 1000000;
+    let ns = (ts % 1000000 * 1000).try_into().unwrap_or_else(|_| {
+        panic!(
+            "Could not get nanoseconds for timestamp {:?} for version {}",
+            ts, version
+        )
+    });
+    chrono::NaiveDateTime::from_timestamp_opt(seconds as i64, ns)
         .unwrap_or_else(|| panic!("Could not parse timestamp {:?} for version {}", ts, version))
 }
 
@@ -189,7 +196,7 @@ pub fn convert_hex(val: String) -> Option<String> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use chrono::Datelike;
+    use chrono::{Datelike, Timelike};
     use serde::Serialize;
 
     #[derive(Serialize, Deserialize, Debug)]
@@ -216,6 +223,7 @@ mod tests {
     fn test_parse_timestamp() {
         let ts = parse_timestamp(1649560602763949, 1);
         assert_eq!(ts.timestamp(), 1649560602);
+        assert_eq!(ts.nanosecond(), 763949000);
         assert_eq!(ts.year(), 2022);
 
         let ts2 = parse_timestamp_secs(600000000000000, 2);

crates/indexer/src/util.rs

@@ -33,6 +33,7 @@ futures = { workspace = true }
 hex = { workspace = true }
 once_cell = { workspace = true }
 prost = { workspace = true }
+regex = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 sha2 = { workspace = true }

ecosystem/indexer-grpc/indexer-grpc-parser/Cargo.toml

@@ -0,0 +1,5 @@
+-- This file should undo anything in `up.sql`\
+DROP TABLE IF EXISTS nft_points;
+DROP INDEX IF EXISTS np_oa_idx;
+DROP INDEX IF EXISTS np_tt_oa_idx;
+DROP INDEX IF EXISTS np_insat_idx;

ecosystem/indexer-grpc/indexer-grpc-parser/migrations/2023-03-08-205402_nft_points/down.sql

@@ -0,0 +1,13 @@
+-- Your SQL goes here
+CREATE TABLE nft_points (
+  transaction_version BIGINT UNIQUE PRIMARY KEY NOT NULL,
+  owner_address VARCHAR(66) NOT NULL,
+  token_name TEXT NOT NULL,
+  point_type TEXT NOT NULL,
+  amount NUMERIC NOT NULL,
+  transaction_timestamp TIMESTAMP NOT NULL,
+  inserted_at TIMESTAMP NOT NULL DEFAULT NOW()
+);
+CREATE INDEX np_oa_idx ON nft_points (owner_address);
+CREATE INDEX np_tt_oa_idx ON nft_points (transaction_timestamp, owner_address);
+CREATE INDEX np_insat_idx ON nft_points (inserted_at);

ecosystem/indexer-grpc/indexer-grpc-parser/migrations/2023-03-08-205402_nft_points/up.sql

@@ -0,0 +1,6 @@
+-- This file should undo anything in `up.sql`
+DROP TABLE IF EXISTS delegated_staking_pools;
+DROP INDEX IF EXISTS dsp_oa_index;
+DROP INDEX IF EXISTS dsp_insat_index;
+ALTER TABLE current_staking_pool_voter
+DROP COLUMN IF EXISTS operator_address;

ecosystem/indexer-grpc/indexer-grpc-parser/migrations/2023-04-02-032121_delegator_pools/down.sql

@@ -0,0 +1,9 @@
+-- Your SQL goes here
+CREATE TABLE IF NOT EXISTS delegated_staking_pools (
+  staking_pool_address VARCHAR(66) UNIQUE PRIMARY KEY NOT NULL,
+  first_transaction_version BIGINT NOT NULL,
+  inserted_at TIMESTAMP NOT NULL DEFAULT NOW()
+);
+CREATE INDEX dsp_insat_index ON delegated_staking_pools (inserted_at);
+ALTER TABLE current_staking_pool_voter
+ADD COLUMN IF NOT EXISTS operator_address VARCHAR(66) NOT NULL;

ecosystem/indexer-grpc/indexer-grpc-parser/migrations/2023-04-02-032121_delegator_pools/up.sql

@@ -0,0 +1,6 @@
+-- This file should undo anything in `up.sql`
+DROP VIEW IF EXISTS address_version_from_events;
+DROP VIEW IF EXISTS address_version_from_move_resources;
+DROP VIEW IF EXISTS current_collection_ownership_view;
+DROP VIEW IF EXISTS num_active_delegator_per_pool;
+DROP INDEX IF EXISTS curr_to_collection_hash_owner_index;