wip: masp

orm/migrations/2025-01-15-130138_transaction_kind_mixed_transfer/down.sql

@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`
+SELECT 1;

orm/migrations/2025-01-15-130138_transaction_kind_mixed_transfer/up.sql

@@ -0,0 +1,2 @@
+-- Your SQL goes here
+ALTER TYPE TRANSACTION_KIND ADD VALUE 'mixed_transfer';

orm/migrations/2025-01-15-135342_masp/down.sql

@@ -0,0 +1,8 @@
+-- This file should undo anything in `up.sql`
+DROP TABLE IF EXISTS masp_pool;
+
+DROP TABLE IF EXISTS masp_pool_aggregate;
+
+DROP TYPE IF EXISTS MASP_POOL_AGGREGATE_WINDOW;
+
+DROP TYPE IF EXISTS MASP_POOL_AGGREGATE_KIND;

orm/migrations/2025-01-15-135342_masp/up.sql

@@ -0,0 +1,139 @@
+-- Your SQL goes here
+CREATE TYPE MASP_POOL_DIRECTION AS ENUM (
+    'in',
+    'out'
+);
+
+CREATE TABLE masp_pool (
+    id SERIAL PRIMARY KEY,
+    token_address VARCHAR(45) NOT NULL,
+    timestamp TIMESTAMP NOT NULL,
+    raw_amount NUMERIC(78, 0) NOT NULL,
+    direction MASP_POOL_DIRECTION NOT NULL,
+    inner_tx_id VARCHAR(64) NOT NULL,
+    CONSTRAINT fk_inner_tx_id FOREIGN KEY(inner_tx_id) REFERENCES inner_transactions(id) ON DELETE CASCADE
+);
+
+CREATE INDEX index_masp_pool_address_timestamp ON masp_pool (token_address, timestamp DESC);
+CREATE UNIQUE INDEX index_masp_pool_inner_tx_id ON masp_pool (inner_tx_id);
+
+CREATE TYPE MASP_POOL_AGGREGATE_WINDOW AS ENUM (
+    'one_day',
+    'seven_days',
+    'thirty_days',
+    'all_time'
+);
+
+CREATE TYPE MASP_POOL_AGGREGATE_KIND AS ENUM (
+    'inflows',
+    'outflows'
+);
+
+CREATE TABLE masp_pool_aggregate (
+    id SERIAL PRIMARY KEY,
+    token_address VARCHAR(45) NOT NULL,
+    time_window MASP_POOL_AGGREGATE_WINDOW NOT NULL,
+    kind MASP_POOL_AGGREGATE_KIND NOT NULL,
+    total_amount NUMERIC(78, 0) NOT NULL DEFAULT 0
+);
+
+CREATE UNIQUE INDEX index_masp_pool_aggregate_token_address_window_kind ON masp_pool_aggregate (token_address, time_window, kind);
+
+
+CREATE OR REPLACE FUNCTION update_masp_pool_aggregate_sum()
+RETURNS TRIGGER AS $$
+DECLARE
+  cutoff_1d TIMESTAMP := now() - INTERVAL '1 day';
+  cutoff_7d TIMESTAMP := now() - INTERVAL '7 days';
+  cutoff_30d TIMESTAMP := now() - INTERVAL '30 days';
+  nk MASP_POOL_AGGREGATE_KIND; -- Declare kind as the ENUM type
+BEGIN
+  -- Determine the kind based on the direction
+  nk := CASE
+            WHEN NEW.direction = 'in' THEN 'inflows'::MASP_POOL_AGGREGATE_KIND
+            ELSE 'outflows'::MASP_POOL_AGGREGATE_KIND
+          END;
+  -- 1 day
+  INSERT INTO masp_pool_aggregate (token_address, time_window, kind, total_amount)
+  VALUES (
+    NEW.token_address,
+    'one_day',
+    nk,
+    (SELECT COALESCE(SUM(raw_amount), 0)
+     FROM masp_pool
+     WHERE token_address = NEW.token_address
+       AND direction = NEW.direction
+       AND timestamp >= cutoff_1d)
+  )
+  ON CONFLICT (token_address, time_window, kind)
+  DO UPDATE SET total_amount = (
+    SELECT COALESCE(SUM(raw_amount), 0)
+    FROM masp_pool
+    WHERE token_address = NEW.token_address
+      AND direction = NEW.direction
+      AND timestamp >= cutoff_1d
+  );
+
+  -- 7 days
+  INSERT INTO masp_pool_aggregate (token_address, time_window, kind, total_amount)
+  VALUES (
+    NEW.token_address,
+    'seven_days',
+    nk,
+    (SELECT COALESCE(SUM(raw_amount), 0)
+     FROM masp_pool
+     WHERE token_address = NEW.token_address
+       AND direction = NEW.direction
+       AND timestamp >= cutoff_1d)
+  )
+  ON CONFLICT (token_address, time_window, kind) 
+  DO UPDATE SET total_amount = (
+    SELECT COALESCE(SUM(raw_amount), 0)
+    FROM masp_pool
+    WHERE token_address = NEW.token_address
+      AND direction = NEW.direction
+      AND timestamp >= cutoff_7d
+  );
+
+  -- 30 days
+  INSERT INTO masp_pool_aggregate (token_address, time_window, kind, total_amount)
+  VALUES (
+    NEW.token_address,
+    'thirty_days',
+    nk,
+    (SELECT COALESCE(SUM(raw_amount), 0)
+     FROM masp_pool
+     WHERE token_address = NEW.token_address
+       AND direction = NEW.direction
+       AND timestamp >= cutoff_1d)
+  )
+  ON CONFLICT (token_address, time_window, kind) 
+  DO UPDATE SET total_amount = (
+    SELECT COALESCE(SUM(raw_amount), 0)
+    FROM masp_pool
+    WHERE token_address = NEW.token_address
+      AND direction = NEW.direction
+      AND timestamp >= cutoff_30d
+  );
+
+  INSERT INTO masp_pool_aggregate (token_address, time_window, kind, total_amount)
+  VALUES (
+    NEW.token_address,
+    'all_time',
+    nk,
+    (SELECT COALESCE(SUM(raw_amount), 0)
+     FROM masp_pool
+     WHERE token_address = NEW.token_address
+       AND direction = NEW.direction)
+  )
+  ON CONFLICT (token_address, time_window, kind) 
+  DO UPDATE SET total_amount = masp_pool_aggregate.total_amount + NEW.raw_amount;
+
+  RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE TRIGGER update_masp_pool_aggregate_sum_trigger
+AFTER INSERT ON masp_pool
+FOR EACH ROW
+EXECUTE FUNCTION update_masp_pool_aggregate_sum();

orm/migrations/2025-01-16-131336_transaction_ibc_types/down.sql

@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`
+SELECT 1;

orm/migrations/2025-01-16-131336_transaction_ibc_types/up.sql

@@ -0,0 +1,4 @@
+-- Your SQL goes here
+ALTER TYPE TRANSACTION_KIND ADD VALUE 'ibc_transparent_transfer';
+ALTER TYPE TRANSACTION_KIND ADD VALUE 'ibc_shielding_transfer';
+ALTER TYPE TRANSACTION_KIND ADD VALUE 'ibc_unshielding_transfer';

orm/src/lib.rs

@@ -8,6 +8,7 @@ pub mod governance_votes;
 pub mod group_by_macros;
 pub mod helpers;
 pub mod ibc;
+pub mod masp;
 pub mod migrations;
 pub mod parameters;
 pub mod pos_rewards;

orm/src/masp.rs

@@ -0,0 +1,76 @@
+use std::str::FromStr;
+
+use bigdecimal::BigDecimal;
+use diesel::{Insertable, Queryable, Selectable};
+use shared::masp::{MaspEntry, MaspEntryDirection};
+
+use crate::schema::{masp_pool, masp_pool_aggregate};
+
+#[derive(Debug, Clone, diesel_derive_enum::DbEnum)]
+#[ExistingTypePath = "crate::schema::sql_types::MaspPoolDirection"]
+pub enum MaspPoolDirectionDb {
+    In,
+    Out,
+}
+
+#[derive(Insertable, Clone, Debug)]
+#[diesel(table_name = masp_pool)]
+#[diesel(check_for_backend(diesel::pg::Pg))]
+pub struct MaspDb {
+    pub token_address: String,
+    pub timestamp: chrono::NaiveDateTime,
+    pub raw_amount: BigDecimal,
+    pub direction: MaspPoolDirectionDb,
+    pub inner_tx_id: String,
+}
+
+pub type MaspInsertDb = MaspDb;
+
+#[derive(Debug, Clone, diesel_derive_enum::DbEnum)]
+#[ExistingTypePath = "crate::schema::sql_types::MaspPoolAggregateWindow"]
+pub enum MaspPoolAggregateWindowDb {
+    OneDay,
+    SevenDays,
+    ThirtyDays,
+    AllTime,
+}
+
+#[derive(Debug, Clone, diesel_derive_enum::DbEnum)]
+#[ExistingTypePath = "crate::schema::sql_types::MaspPoolAggregateKind"]
+pub enum MaspPoolAggregateKindDb {
+    Inflows,
+    Outflows,
+}
+
+#[derive(Insertable, Queryable, Selectable, Clone, Debug)]
+#[diesel(table_name = masp_pool_aggregate)]
+#[diesel(check_for_backend(diesel::pg::Pg))]
+pub struct MaspPoolDb {
+    pub id: i32,
+    pub token_address: String,
+    pub time_window: MaspPoolAggregateWindowDb,
+    pub kind: MaspPoolAggregateKindDb,
+    pub total_amount: BigDecimal,
+}
+
+impl From<MaspEntry> for MaspInsertDb {
+    fn from(value: MaspEntry) -> Self {
+        let timestamp = chrono::DateTime::from_timestamp(value.timestamp, 0)
+            .expect("Invalid timestamp")
+            .naive_utc();
+
+        let amount = BigDecimal::from_str(&value.raw_amount.to_string())
+            .expect("Invalid amount");
+
+        MaspInsertDb {
+            token_address: value.token_address,
+            timestamp,
+            raw_amount: amount,
+            direction: match value.direction {
+                MaspEntryDirection::In => MaspPoolDirectionDb::In,
+                MaspEntryDirection::Out => MaspPoolDirectionDb::Out,
+            },
+            inner_tx_id: value.inner_tx_id.to_string(),
+        }
+    }
+}

orm/src/schema.rs

@@ -49,6 +49,30 @@ pub mod sql_types {
     #[diesel(postgres_type(name = "ibc_status"))]
     pub struct IbcStatus;
 
+    #[derive(
+        diesel::query_builder::QueryId,
+        std::fmt::Debug,
+        diesel::sql_types::SqlType,
+    )]
+    #[diesel(postgres_type(name = "masp_pool_aggregate_kind"))]
+    pub struct MaspPoolAggregateKind;
+
+    #[derive(
+        diesel::query_builder::QueryId,
+        std::fmt::Debug,
+        diesel::sql_types::SqlType,
+    )]
+    #[diesel(postgres_type(name = "masp_pool_aggregate_window"))]
+    pub struct MaspPoolAggregateWindow;
+
+    #[derive(
+        diesel::query_builder::QueryId,
+        std::fmt::Debug,
+        diesel::sql_types::SqlType,
+    )]
+    #[diesel(postgres_type(name = "masp_pool_direction"))]
+    pub struct MaspPoolDirection;
+
     #[derive(
         diesel::query_builder::QueryId,
         std::fmt::Debug,
@@ -268,6 +292,37 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    use diesel::sql_types::*;
+    use super::sql_types::MaspPoolDirection;
+
+    masp_pool (id) {
+        id -> Int4,
+        #[max_length = 45]
+        token_address -> Varchar,
+        timestamp -> Timestamp,
+        raw_amount -> Numeric,
+        direction -> MaspPoolDirection,
+        #[max_length = 64]
+        inner_tx_id -> Varchar,
+    }
+}
+
+diesel::table! {
+    use diesel::sql_types::*;
+    use super::sql_types::MaspPoolAggregateWindow;
+    use super::sql_types::MaspPoolAggregateKind;
+
+    masp_pool_aggregate (id) {
+        id -> Int4,
+        #[max_length = 45]
+        token_address -> Varchar,
+        time_window -> MaspPoolAggregateWindow,
+        kind -> MaspPoolAggregateKind,
+        total_amount -> Numeric,
+    }
+}
+
 diesel::table! {
     pos_rewards (id) {
         id -> Int4,
@@ -363,6 +418,7 @@ diesel::joinable!(gas_estimations -> wrapper_transactions (wrapper_id));
 diesel::joinable!(governance_votes -> governance_proposals (proposal_id));
 diesel::joinable!(ibc_token -> token (address));
 diesel::joinable!(inner_transactions -> wrapper_transactions (wrapper_id));
+diesel::joinable!(masp_pool -> inner_transactions (inner_tx_id));
 diesel::joinable!(pos_rewards -> validators (validator_id));
 diesel::joinable!(transaction_history -> inner_transactions (inner_tx_id));
 diesel::joinable!(unbonds -> validators (validator_id));
@@ -382,6 +438,8 @@ diesel::allow_tables_to_appear_in_same_query!(
     ibc_ack,
     ibc_token,
     inner_transactions,
+    masp_pool,
+    masp_pool_aggregate,
     pos_rewards,
     revealed_pk,
     token,

orm/src/transactions.rs

@@ -16,7 +16,11 @@ pub enum TransactionKindDb {
     ShieldedTransfer,
     ShieldingTransfer,
     UnshieldingTransfer,
+    MixedTransfer,
     IbcMsgTransfer,
+    IbcTransparentTransfer,
+    IbcShieldingTransfer,
+    IbcUnshieldingTransfer,
     Bond,
     Redelegation,
     Unbond,
@@ -41,7 +45,21 @@ impl From<TransactionKind> for TransactionKindDb {
                 Self::TransparentTransfer
             }
             TransactionKind::ShieldedTransfer(_) => Self::ShieldedTransfer,
+            TransactionKind::UnshieldingTransfer(_) => {
+                Self::UnshieldingTransfer
+            }
+            TransactionKind::ShieldingTransfer(_) => Self::ShieldingTransfer,
+            TransactionKind::MixedTransfer(_) => Self::MixedTransfer,
             TransactionKind::IbcMsgTransfer(_) => Self::IbcMsgTransfer,
+            TransactionKind::IbcTrasparentTransfer(_) => {
+                Self::IbcTransparentTransfer
+            }
+            TransactionKind::IbcShieldingTransfer(_) => {
+                Self::IbcShieldingTransfer
+            }
+            TransactionKind::IbcUnshieldingTransfer(_) => {
+                Self::IbcUnshieldingTransfer
+            }
             TransactionKind::Bond(_) => Self::Bond,
             TransactionKind::Redelegation(_) => Self::Redelegation,
             TransactionKind::Unbond(_) => Self::Unbond,