[V1] Transaction Pipe Refactor

crates/core/src/instruction.rs

@@ -10,15 +10,6 @@ use {
     },
 };
 
-use crate::collection::InstructionDecoderCollection;
-
-#[derive(Debug)]
-pub struct ParsedInstruction<T: InstructionDecoderCollection> {
-    pub metadata: InstructionMetadata,
-    pub instruction: T,
-    pub inner_instructions: Vec<ParsedInstruction<T>>,
-}
-
 #[derive(Debug, Clone)]
 pub struct InstructionMetadata {
     pub transaction_metadata: Arc<TransactionMetadata>,

crates/core/src/pipeline.rs

@@ -337,7 +337,8 @@ impl Pipeline {
                         &transaction_update,
                     )?;
 
-                let nested_instructions: NestedInstructions = instructions_with_metadata.into();
+                let nested_instructions: NestedInstructions =
+                    instructions_with_metadata.clone().into();
                 let mut all_instructions = Vec::new();
                 flatten_nested_instructions(&nested_instructions, &mut all_instructions);
 
@@ -369,7 +370,7 @@ impl Pipeline {
                             FilterResult::Accept
                         )
                     }) {
-                        pipe.run(transaction_metadata.clone(), &nested_instructions)
+                        pipe.run(transaction_metadata.clone(), &instructions_with_metadata)
                             .await?;
                     }
                 }

crates/core/src/transaction.rs

@@ -3,14 +3,12 @@ use solana_program::hash::Hash;
 
 use {
     crate::{
-        collection::InstructionDecoderCollection,
-        error::CarbonResult,
-        instruction::{InstructionMetadata, NestedInstruction, ParsedInstruction},
-        processor::Processor,
-        transformers,
+        collection::InstructionDecoderCollection, error::CarbonResult,
+        instruction::InstructionMetadata, processor::Processor,
     },
     async_trait::async_trait,
     core::convert::TryFrom,
+    solana_instruction::Instruction,
     solana_pubkey::Pubkey,
     solana_signature::Signature,
     std::sync::Arc,
@@ -61,11 +59,6 @@ pub struct TransactionPipe<T: InstructionDecoderCollection, P> {
     _phantom: std::marker::PhantomData<T>,
 }
 
-pub struct ParsedTransaction<I: InstructionDecoderCollection> {
-    pub metadata: TransactionMetadata,
-    pub instructions: Vec<ParsedInstruction<I>>,
-}
-
 impl<T: InstructionDecoderCollection, P> TransactionPipe<T, P> {
     pub fn new(processor: P, filters: Vec<Box<dyn Filter + Send + Sync + 'static>>) -> Self {
         log::trace!(
@@ -80,36 +73,28 @@ impl<T: InstructionDecoderCollection, P> TransactionPipe<T, P> {
     }
 }
 
-pub fn parse_instructions<T: InstructionDecoderCollection>(
-    nested_ixs: &[NestedInstruction],
-) -> Vec<ParsedInstruction<T>> {
-    log::trace!("parse_instructions(nested_ixs: {nested_ixs:?})");
-
-    let mut parsed_instructions: Vec<ParsedInstruction<T>> = Vec::new();
-
-    for nested_ix in nested_ixs {
-        if let Some(instruction) = T::parse_instruction(&nested_ix.instruction) {
-            parsed_instructions.push(ParsedInstruction {
-                metadata: nested_ix.metadata.clone(),
-                instruction,
-                inner_instructions: parse_instructions(&nested_ix.inner_instructions),
-            });
-        } else {
-            for inner_ix in nested_ix.inner_instructions.iter() {
-                parsed_instructions.extend(parse_instructions(std::slice::from_ref(inner_ix)));
-            }
-        }
-    }
-
-    parsed_instructions
+pub fn parse_instructions_flat<T: InstructionDecoderCollection>(
+    instructions: &[(InstructionMetadata, Instruction)],
+) -> Vec<(InstructionMetadata, T)> {
+    log::trace!(
+        "parse_instructions_flat(instructions: len={})",
+        instructions.len()
+    );
+
+    instructions
+        .iter()
+        .filter_map(|(metadata, instruction)| {
+            T::parse_instruction(instruction).map(|parsed| (metadata.clone(), parsed))
+        })
+        .collect()
 }
 
 #[async_trait]
 pub trait TransactionPipes<'a>: Send + Sync {
     async fn run(
         &mut self,
         transaction_metadata: Arc<TransactionMetadata>,
-        instructions: &[NestedInstruction],
+        instructions: &[(InstructionMetadata, Instruction)],
     ) -> CarbonResult<()>;
 
     fn filters(&self) -> &Vec<Box<dyn Filter + Send + Sync + 'static>>;
@@ -124,17 +109,18 @@ where
     async fn run(
         &mut self,
         transaction_metadata: Arc<TransactionMetadata>,
-        instructions: &[NestedInstruction],
+        instructions: &[(InstructionMetadata, Instruction)],
     ) -> CarbonResult<()> {
-        log::trace!("TransactionPipe::run(instructions: {instructions:?})");
-
-        let parsed_instructions = parse_instructions(instructions);
+        log::trace!(
+            "TransactionPipe::run(instructions: len={})",
+            instructions.len()
+        );
 
-        let unnested_instructions = transformers::unnest_parsed_instructions(parsed_instructions);
+        let parsed_instructions = parse_instructions_flat::<T>(instructions);
 
         let data = TransactionProcessorInputType {
             metadata: &transaction_metadata,
-            instructions: &unnested_instructions,
+            instructions: &parsed_instructions,
         };
 
         self.processor.process(&data).await?;

crates/core/src/transformers.rs

@@ -1,9 +1,8 @@
 use {
     crate::{
-        collection::InstructionDecoderCollection,
         datasource::TransactionUpdate,
         error::{CarbonResult, Error},
-        instruction::{InstructionMetadata, ParsedInstruction, MAX_INSTRUCTION_STACK_DEPTH},
+        instruction::{InstructionMetadata, MAX_INSTRUCTION_STACK_DEPTH},
         transaction::TransactionMetadata,
     },
     solana_instruction::AccountMeta,
@@ -195,23 +194,6 @@ pub fn extract_account_metas(
     Ok(accounts)
 }
 
-pub fn unnest_parsed_instructions<T: InstructionDecoderCollection>(
-    instructions: Vec<ParsedInstruction<T>>,
-) -> Vec<(InstructionMetadata, T)> {
-    log::trace!("unnest_parsed_instructions(instructions: {instructions:?})");
-
-    let mut result = Vec::new();
-
-    for parsed_instruction in instructions {
-        result.push((parsed_instruction.metadata, parsed_instruction.instruction));
-        result.extend(unnest_parsed_instructions(
-            parsed_instruction.inner_instructions,
-        ));
-    }
-
-    result
-}
-
 pub fn transaction_metadata_from_original_meta(
     meta_original: UiTransactionStatusMeta,
 ) -> CarbonResult<TransactionStatusMeta> {

crates/macros/src/lib.rs

@@ -1,4 +1,3 @@
 #![no_std]
 
-pub mod schemas;
 pub mod try_decode_ixs;