[Ecosystem][Indexer] Add transaction WriteSet support to Indexer

api/types/src/transaction.rs

@@ -475,6 +475,17 @@ pub enum WriteSetChange {
     },
 }
 
+impl WriteSetChange {
+    pub fn type_str(&self) -> &'static str {
+        match self {
+            WriteSetChange::DeleteModule { .. } => "delete_module",
+            WriteSetChange::DeleteResource { .. } => "delete_resource",
+            WriteSetChange::WriteModule { .. } => "write_module",
+            WriteSetChange::WriteResource { .. } => "write_resource",
+        }
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(tag = "type", rename_all = "snake_case")]
 pub enum TransactionSignature {

ecosystem/indexer/migrations/2022-04-19-003410_create_write_set_changes/down.sql

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

ecosystem/indexer/migrations/2022-04-19-003410_create_write_set_changes/up.sql

@@ -0,0 +1,25 @@
+-- Your SQL goes here
+
+CREATE TABLE write_set_changes
+(
+    -- join from "transactions"
+    transaction_hash VARCHAR(255) NOT NULL,
+
+    hash             VARCHAR(255) NOT NULL,
+
+    type             TEXT         NOT NULL,
+    address          VARCHAR(255) NOT NULL,
+
+    module           jsonb        NOT NULL,
+    resource         jsonb        NOT NULL,
+    data             jsonb        NOT NULL,
+    inserted_at      TIMESTAMP    NOT NULL DEFAULT NOW(),
+
+    -- Constraints
+    PRIMARY KEY (transaction_hash, hash),
+    CONSTRAINT fk_transactions
+        FOREIGN KEY (transaction_hash)
+            REFERENCES transactions (hash)
+);
+
+CREATE INDEX write_set_changes_tx_hash_addr_type_index ON write_set_changes (transaction_hash, address, type);

ecosystem/indexer/src/default_processor.rs

@@ -10,6 +10,7 @@ use crate::{
     models::{
         events::EventModel,
         transactions::{BlockMetadataTransactionModel, TransactionModel, UserTransactionModel},
+        write_set_changes::WriteSetChangeModel,
     },
     schema,
 };
@@ -50,6 +51,16 @@ fn insert_events(conn: &PgPoolConnection, events: &Vec<EventModel>) {
     .expect("Error inserting row into database");
 }
 
+fn insert_write_set_changes(conn: &PgPoolConnection, write_set_changes: &Vec<WriteSetChangeModel>) {
+    execute_with_better_error(
+        conn,
+        diesel::insert_into(schema::write_set_changes::table)
+            .values(write_set_changes)
+            .on_conflict_do_nothing(),
+    )
+    .expect("Error inserting row into database");
+}
+
 fn insert_transaction(conn: &PgPoolConnection, version: u64, transaction_model: &TransactionModel) {
     aptos_logger::trace!(
         "[default_processor] inserting 'transaction' version {} with hash {}",
@@ -123,7 +134,7 @@ impl TransactionProcessor for DefaultTransactionProcessor {
     ) -> Result<ProcessingResult, TransactionProcessingError> {
         let version = transaction.version().unwrap_or(0);
 
-        let (transaction_model, maybe_details_model, maybe_events) =
+        let (transaction_model, maybe_details_model, maybe_events, maybe_write_set_changes) =
             TransactionModel::from_transaction(&transaction);
 
         let conn = self.get_conn();
@@ -154,6 +165,9 @@ impl TransactionProcessor for DefaultTransactionProcessor {
             if let Some(events) = maybe_events {
                 insert_events(&conn, &events);
             };
+            if let Some(write_set_changes) = maybe_write_set_changes {
+                insert_write_set_changes(&conn, &write_set_changes);
+            };
             Ok(())
         });