marketplace indexer (#4)

* nit

* define all tables

* fix migration

* only have 2 tables

* done with marketplace db schema

* storer wip

* save

* time to test non tradeport indexer

* nit

* tradeport wip

* skip tls validation

* Update .gitignore

* nit

* fix event struct

* log

* fix

* nit

* dedup order placed events

* debug

* Update ask_filled_event_storer.rs

* debug

* debug

* Update ask_filled_event_storer.rs

* debug

* log

* refactor wip

* use default vaule when not writing to db

* fmt

* fix default value

* wip

* added all tradeport events

* parse tradeport

* nit

* nit

indexer/.gitignore

@@ -21,4 +21,4 @@ local.config.yaml
 cloud.config.yaml
 config.yaml
 contract_upgrade_indexer_config.yaml
-marketplace_indexer_config.yaml
+*_marketplace_indexer_config.yaml

indexer/Cargo.toml

@@ -16,6 +16,7 @@ aptos-indexer-processor-sdk-server-framework = { git = "https://github.com/aptos
 ahash = { version = "0.8.7", features = ["serde"] }
 anyhow = "1.0.86"
 async-trait = "0.1.80"
+blake3 = "1.5.5"
 chrono = { version = "0.4.19", features = ["clock", "serde"] }
 clap = { version = "4.3.5", features = ["derive", "unstable-styles"] }
 # Do NOT enable the postgres feature here, it is conditionally enabled in a feature
@@ -38,6 +39,7 @@ diesel-async = { git = "https://github.com/weiznich/diesel_async.git", rev = "d0
 diesel_migrations = { version = "2.1.0", features = ["postgres"] }
 field_count = "0.1.1"
 futures-util = "0.3.21"
+hex = "0.4.3"
 jemallocator = { version = "0.5.0", features = [
     "profiling",
     "unprefixed_malloc_on_supported_platforms",

indexer/Dockerfile

@@ -39,10 +39,11 @@ WORKDIR /usr/src/app
 COPY --from=builder /usr/src/app/target/x86_64-unknown-linux-gnu/release/indexer .
 
 # Copy the configuration file
-COPY configs/contract_upgrade_indexer_config.yaml /secrets/config
+COPY configs/contract_upgrade_indexer_config.yaml /secrets/contract_upgrade_config
+COPY configs/marketplace_indexer_config.yaml /secrets/marketplace_config
 
 # Expose the port your application is using
 EXPOSE 8080
 
-# Set the command to run the application
-CMD ["./indexer", "-c", "/secrets/config"]
+# Set the binary as entrypoint so we can pass config as argument
+ENTRYPOINT ["./indexer", "-c"]

indexer/configs/example.contract_upgrade_indexer_config.yaml

@@ -4,6 +4,9 @@ server_config:
     type: "contract_upgrade_indexer"
   transaction_stream_config:
     indexer_grpc_data_service_address: "https://grpc.mainnet.aptoslabs.com:443"
+    # rarible starting version: 1011760686
+    # wapal starting version:    216804176
+    # tradeport starting version: 94194505
     starting_version: 1
     # request_ending_version: 10000
     auth_token: "_"

indexer/configs/example.marketplace_indexer_config.yaml

@@ -4,6 +4,9 @@ server_config:
     type: "marketplace_indexer"
   transaction_stream_config:
     indexer_grpc_data_service_address: "https://grpc.mainnet.aptoslabs.com:443"
+    # rarible starting version: 1011760686
+    # wapal starting version:    216804176
+    # tradeport starting version: 94194505
     starting_version: 1
     # request_ending_version: 10000
     auth_token: ""
@@ -16,7 +19,7 @@ server_config:
     # see limitation on vercel docs https://vercel.com/docs/storage/vercel-postgres/faq
     db_pool_size: 25
   custom_config:
-    contract_upgrade_indexer: [
+    marketplace_indexer: [
         # wapal
         "0x584b50b999c78ade62f8359c91b5165ff390338d45f8e55969a04e65d76258c9",
         # tradeport

indexer/src/config/indexer_processor_config.rs

@@ -4,7 +4,10 @@ use aptos_indexer_processor_sdk_server_framework::RunnableConfig;
 use serde::{Deserialize, Serialize};
 
 use super::processor_config::ProcessorConfig;
-use crate::indexers::contract_upgrade_indexer::processor::ContractUpgradeProcessor;
+use crate::indexers::{
+    contract_upgrade_indexer::processor::ContractUpgradeProcessor,
+    marketplace_indexer::processor::MarketplaceProcessor,
+};
 
 pub const QUERY_DEFAULT_RETRIES: u32 = 5;
 pub const QUERY_DEFAULT_RETRY_DELAY_MS: u64 = 500;
@@ -23,11 +26,20 @@ impl RunnableConfig for IndexerProcessorConfig {
     async fn run(&self) -> Result<()> {
         match self.processor_config {
             ProcessorConfig::ContractUpgradeIndexer => {
-                let events_processor = ContractUpgradeProcessor::new(self.clone()).await?;
-                events_processor.run_processor().await
+                let processor = ContractUpgradeProcessor::new(self.clone()).await?;
+                processor.run_processor().await
             }
-            ProcessorConfig::MarketplaceIndexer => {
-                return Err(anyhow::anyhow!("MarketplaceIndexer not implemented"));
+            ProcessorConfig::WapalMarketplaceIndexer => {
+                let processor = MarketplaceProcessor::new(self.clone()).await?;
+                processor.run_processor().await
+            }
+            ProcessorConfig::RaribleMarketplaceIndexer => {
+                let processor = MarketplaceProcessor::new(self.clone()).await?;
+                processor.run_processor().await
+            }
+            ProcessorConfig::TradeportMarketplaceIndexer => {
+                let processor = MarketplaceProcessor::new(self.clone()).await?;
+                processor.run_processor().await
             }
         }
     }

indexer/src/config/processor_config.rs

@@ -33,7 +33,9 @@ use serde::{Deserialize, Serialize};
 )]
 pub enum ProcessorConfig {
     ContractUpgradeIndexer,
-    MarketplaceIndexer,
+    RaribleMarketplaceIndexer,
+    WapalMarketplaceIndexer,
+    TradeportMarketplaceIndexer,
 }
 
 impl ProcessorConfig {
@@ -59,7 +61,9 @@ impl ProcessorConfig {
 )]
 pub enum Processor {
     ContractUpgradeIndexer,
-    MarketplaceIndexer,
+    RaribleMarketplaceIndexer,
+    WapalMarketplaceIndexer,
+    TradeportMarketplaceIndexer,
 }
 
 #[cfg(test)]

indexer/src/db_migrations/contract_upgrade_indexer/diesel.toml → indexer/src/db_migrations/diesel.toml

@@ -2,7 +2,7 @@
 # see https://diesel.rs/guides/configuring-diesel-cli
 
 [print_schema]
-file = "contract_upgrade_schema.rs"
+file = "schema.rs"
 
 [migrations_directory]
 dir = "migrations"

indexer/src/db_migrations/migrations/2024-07-18-194547_create-processor-status/down.sql

@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`d
+DROP TABLE IF EXISTS processor_status;

indexer/src/db_migrations/contract_upgrade_indexer/migrations/2024-07-18-194547_create-processor-status/up.sql → indexer/src/db_migrations/migrations/2024-07-18-194547_create-processor-status/up.sql

@@ -1,6 +1,6 @@
 -- Your SQL goes here
 CREATE TABLE
-    contract_upgrade_processor_status (
+    processor_status (
         processor VARCHAR(50) NOT NULL,
         last_success_version BIGINT NOT NULL,
         last_updated TIMESTAMP NOT NULL,

indexer/src/db_migrations/contract_upgrade_indexer/migrations/2024-07-18-202400_test-migration/up.sql → indexer/src/db_migrations/migrations/2024-07-18-202400_test-migration/up.sql

@@ -1,4 +1,4 @@
 -- Your SQL goes here
-ALTER TABLE IF EXISTS contract_upgrade_processor_status
+ALTER TABLE IF EXISTS processor_status
 ALTER COLUMN last_updated
 SET DEFAULT NOW ();

indexer/src/db_migrations/migrations/2024-08-01-224558_ledger-infos/down.sql

@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`
+DROP TABLE IF EXISTS ledger_infos;

indexer/src/db_migrations/migrations/2024-08-01-224558_ledger-infos/up.sql

@@ -0,0 +1,3 @@
+-- Your SQL goes here
+CREATE TABLE
+    ledger_infos (chain_id BIGINT UNIQUE PRIMARY KEY NOT NULL);

indexer/src/db_migrations/migrations/2024-12-03-235031_add-nft-asks-table/down.sql

@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`d
+DROP TABLE IF EXISTS nft_asks;

indexer/src/db_migrations/migrations/2024-12-03-235031_add-nft-asks-table/up.sql

@@ -0,0 +1,47 @@
+-- This table is backfill safe, i.e. you can re-index without dropping the table
+CREATE TABLE
+    nft_asks (
+        ask_obj_addr VARCHAR(300) PRIMARY KEY,
+        -- For v1 NFTs, this is property_version, for v2 NFTs, this is nft_addr
+        nft_id VARCHAR(300) NOT NULL,
+        nft_name VARCHAR(300) NOT NULL,
+        -- For v2 NFTs, we use collection_addr to identify the collection
+        collection_addr VARCHAR(300) NOT NULL,
+        -- For v1 NFTs, we use creator_addr + name to identify the collection
+        collection_creator_addr VARCHAR(300) NOT NULL,
+        collection_name VARCHAR(300) NOT NULL,
+        -- 1 is token v1, 2 is token v2
+        nft_standard INT NOT NULL,
+        marketplace_addr VARCHAR(300) NOT NULL,
+        -- empty str when it's not filled
+        buyer_addr VARCHAR(300) NOT NULL,
+        seller_addr VARCHAR(300) NOT NULL,
+        -- price in on-chain unit, for APT it's oct
+        price BIGINT NOT NULL,
+        -- in on-chain unit, for APT it's oct
+        royalties BIGINT NOT NULL,
+        -- in on-chain unit, for APT it's oct
+        commission BIGINT NOT NULL,
+        -- for coin APT, this is 0x1::aptos_coin::AptosCoin
+        -- for fa APT, this is 0xa
+        payment_token VARCHAR(300) NOT NULL,
+        -- 1 is coin, 2 is fa
+        payment_token_type INT NOT NULL,
+        order_placed_timestamp BIGINT NOT NULL,
+        order_placed_tx_version BIGINT NOT NULL,
+        order_placed_event_idx BIGINT NOT NULL,
+        order_filled_timestamp BIGINT NOT NULL,
+        order_filled_tx_version BIGINT NOT NULL,
+        order_filled_event_idx BIGINT NOT NULL,
+        order_cancelled_timestamp BIGINT NOT NULL,
+        order_cancelled_tx_version BIGINT NOT NULL,
+        order_cancelled_event_idx BIGINT NOT NULL,
+        -- 1 is active, 2 is filled, 3 is cancelled
+        order_status INT NOT NULL,
+        -- 1 is fixed price, 2 is auction
+        order_type INT NOT NULL,
+        CHECK (nft_standard IN (1, 2)),
+        CHECK (payment_token_type IN (1, 2)),
+        CHECK (order_status IN (1, 2, 3)),
+        CHECK (order_type IN (1, 2))
+    );

indexer/src/db_migrations/migrations/2024-12-10-014414_add-index-to-nft-asks-table/down.sql

@@ -0,0 +1,16 @@
+-- This file should undo anything in `up.sql`
+DROP INDEX idx_nft_asks_nft_id;
+
+DROP INDEX idx_nft_asks_nft_name;
+
+DROP INDEX idx_nft_asks_collection_addr;
+
+DROP INDEX idx_nft_asks_collection_creator_addr;
+
+DROP INDEX idx_nft_asks_collection_name;
+
+DROP INDEX idx_nft_asks_buyer_addr;
+
+DROP INDEX idx_nft_asks_seller_addr;
+
+DROP INDEX idx_nft_asks_marketplace_addr;