Record transactions into pindexer

crates/bin/pindexer/src/dex_ex/mod.rs

@@ -703,6 +703,8 @@ struct Events {
     position_open_txs: BTreeMap<PositionId, [u8; 32]>,
     position_close_txs: BTreeMap<PositionId, [u8; 32]>,
     position_withdrawal_txs: BTreeMap<PositionId, [u8; 32]>,
+    // Track transactions
+    transactions: HashMap<TransactionId, Transaction>,
 }
 
 impl Events {
@@ -723,6 +725,7 @@ impl Events {
             position_open_txs: BTreeMap::new(),
             position_close_txs: BTreeMap::new(),
             position_withdrawal_txs: BTreeMap::new(),
+            transactions: HashMap::new(),
         }
     }
 
@@ -888,6 +891,8 @@ impl Events {
                     .entry(e.trading_pair)
                     .or_insert_with(Vec::new)
                     .push(e);
+            } else if let Ok(e) = EventBlockTransaction::try_from_event(&event.event) {
+                out.transactions.insert(e.transaction_id, e.transaction);
             }
         }
         Ok(out)
@@ -1397,6 +1402,32 @@ impl Component {
 
         Ok(())
     }
+
+    async fn record_transaction(
+        &self,
+        dbtx: &mut PgTransaction<'_>,
+        time: DateTime,
+        height: i32,
+        transaction_id: TransactionId,
+        transaction: Transaction,
+    ) -> anyhow::Result<()> {
+        sqlx::query(
+            "INSERT INTO dex_ex_transactions (
+                transaction_id,
+                transaction,
+                height,
+                time
+            ) VALUES ($1, $2, $3, $4)",
+        )
+        .bind(transaction_id)
+        .bind(transaction)
+        .bind(height)
+        .bind(time)
+        .execute(dbtx.as_mut())
+        .await?;
+
+        Ok(())
+    }
 }
 
 #[async_trait]
@@ -1475,6 +1506,12 @@ impl AppView for Component {
                     .await?;
             }
 
+            // Record transactions
+            for (transaction_id, transaction) in &events.transactions {
+                self.record_transaction(dbtx, time, events.height, transaction_id, transaction)
+                    .await?;
+            }
+
             for (pair, candle) in &events.candles {
                 for window in Window::all() {
                     let key = (pair.start, pair.end, window);

crates/bin/pindexer/src/dex_ex/schema.sql

@@ -251,6 +251,22 @@ CREATE TABLE IF NOT EXISTS dex_ex_block_summary (
 
 CREATE INDEX ON dex_ex_block_summary (time, height);
 
+CREATE TABLE IF NOT EXISTS dex_ex_transactions (
+  -- The primary key
+  rowid SERIAL PRIMARY KEY,
+  -- The unique identifier of the transaction
+  transaction_id BYTEA NOT NULL UNIQUE,
+  -- The raw transaction bytes
+  transaction BYTEA NOT NULL,
+  -- The block height at which this transaction was included
+  height INTEGER NOT NULL,
+  -- The timestamp when this transaction was included in a block
+  time TIMESTAMPTZ NOT NULL,
+  PRIMARY KEY (transaction_id)
+);
+
+CREATE INDEX ON dex_ex_transactions (time, height);
+
 ALTER TABLE dex_ex_position_executions
   ADD CONSTRAINT fk_position_executions
   FOREIGN KEY (position_id) REFERENCES dex_ex_position_state(position_id);

crates/core/component/dex/src/event.rs

@@ -661,3 +661,42 @@ impl From<EventCandlestickData> for pb::EventCandlestickData {
 impl DomainType for EventCandlestickData {
     type Proto = pb::EventCandlestickData;
 }
+
+#[derive(Clone, Debug)]
+pub struct EventBlockTransaction {
+    pub transaction_id: TransactionId,
+    pub transaction: Transaction,
+}
+
+impl TryFrom<pb::EventBlockTransaction> for EventBlockTransaction {
+    type Error = anyhow::Error;
+
+    fn try_from(value: pb::EventBlockTransaction) -> Result<Self, Self::Error> {
+        fn inner(value: pb::EventBlockTransaction) -> anyhow::Result<EventBlockTransaction> {
+            Ok(EventBlockTransaction {
+                transaction_id: value
+                    .transaction_id
+                    .ok_or(anyhow!("missing `transaction_id`"))?
+                    .try_into()?,
+                transaction: value
+                    .transaction
+                    .ok_or(anyhow!("missing `transaction`"))?
+                    .try_into()?,
+            })
+        }
+        inner(value).context(format!("parsing {}", pb::EventBlockTransaction::NAME))
+    }
+}
+
+impl From<EventBlockTransaction> for pb::EventBlockTransaction {
+    fn from(value: EventBlockTransaction) -> Self {
+        Self {
+            transaction_id: Some(value.transaction_id.into()),
+            transaction: Some(value.transaction.into()),
+        }
+    }
+}
+
+impl DomainType for EventBlockTransaction {
+    type Proto = pb::EventBlockTransaction;
+}