chore: extract transaction syncer API

.github/workflows/ci.yml

@@ -0,0 +1,49 @@
+on:
+  pull_request:
+    types: [opened, reopened, synchronize, ready_for_review]
+
+name: CI
+
+jobs:
+  ci:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Setup Rust stable
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          components: rustfmt, clippy
+
+      - name: Setup Rust nightly for formatting
+        uses: dtolnay/rust-toolchain@nightly
+        with:
+          components: rustfmt, clippy
+
+      - name: Cache cargo registry
+        uses: actions/cache@v4
+        with:
+          path: ~/.cargo/registry
+          key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }}
+
+      - name: Cache cargo index
+        uses: actions/cache@v4
+        with:
+          path: ~/.cargo/git
+          key: ${{ runner.os }}-cargo-index-${{ hashFiles('**/Cargo.lock') }}
+
+      - name: Cache cargo build
+        uses: actions/cache@v4
+        with:
+          path: target
+          key: ${{ runner.os }}-cargo-build-target-${{ hashFiles('**/Cargo.lock') }}
+
+      - name: Run tests
+        run: cargo test
+
+      - name: Check formatting
+        run: cargo +nightly fmt --check
+
+      - name: Run clippy
+        run: cargo clippy --all-targets -- -D warnings

Cargo.toml

@@ -6,6 +6,8 @@ description = "Real-time synchronization of magicblock delegation records"
 readme = "README.md"
 
 [dependencies]
+bs58 = "0.5.1"
+either = "1.15.0"
 futures = "0.3"
 helius-laserstream = "0.1.5"
 tokio = { version = "1.0", features = ["sync", "macros"] }

src/consts.rs

@@ -0,0 +1,34 @@
+//! Constants used throughout the DLP synchronization service.
+
+use crate::types::Pubkey;
+
+/// Size of a Solana public key in bytes.
+pub(crate) const PUBKEY_LEN: usize = 32;
+
+/// Delegation program address.
+pub(crate) const DELEGATION_PROGRAM: &str = "DELeGGvXpWV2fqJUhqcF5ZSYMS4JTLjteaAMARRSaeSh";
+
+/// Delegation program pubkey in bytes.
+pub(crate) const DELEGATION_PROGRAM_PUBKEY: Pubkey =
+    bs58::decode(DELEGATION_PROGRAM.as_bytes()).into_array_const_unwrap();
+
+/// Size of a delegation record account in bytes.
+pub(crate) const DELEGATION_RECORD_SIZE: u64 = 96;
+
+/// Instruction discriminator for undelegate operations.
+pub(crate) const UNDELEGATE_DISCRIMINATOR: u8 = 3;
+
+/// Length of an instruction discriminator (Anchor programs).
+pub(crate) const DISCRIMINATOR_LEN: usize = 8;
+
+/// Index of the delegation record account in undelegate instruction accounts.
+pub(crate) const DELEGATION_RECORD_ACCOUNT_INDEX: usize = 7;
+
+/// Maximum pending subscription/unsubscription requests.
+pub(crate) const MAX_PENDING_REQUESTS: usize = 256;
+
+/// Maximum pending account/transaction updates.
+pub(crate) const MAX_PENDING_UPDATES: usize = 8192;
+
+/// Maximum reconnection attempts to the Laserstream.
+pub(crate) const MAX_RECONNECT_ATTEMPTS: u32 = 16;

src/lib.rs

@@ -5,7 +5,7 @@
 //!
 //! # Usage
 //!
-//! ```no_run
+//! ```ignore
 //! use dlp_sync::DlpSyncer;
 //!
 //! # async fn example() -> Result<(), dlp_sync::DlpSyncError> {
@@ -39,7 +39,9 @@
 //! ```
 
 mod channels;
+mod consts;
 mod syncer;
+pub mod transaction_syncer;
 mod types;
 
 pub use channels::{DlpSyncChannelsInit, DlpSyncChannelsRequester};

src/syncer.rs

@@ -10,10 +10,8 @@ use helius_laserstream::{
     grpc::{
         subscribe_request_filter_accounts_filter::Filter, subscribe_update::UpdateOneof,
         SubscribeRequest, SubscribeRequestFilterAccounts, SubscribeRequestFilterAccountsFilter,
-        SubscribeRequestFilterTransactions, SubscribeRequestPing, SubscribeUpdate,
-        SubscribeUpdateAccount, SubscribeUpdateTransaction,
+        SubscribeRequestPing, SubscribeUpdate, SubscribeUpdateAccount, SubscribeUpdateTransaction,
     },
-    solana::storage::confirmed_block::CompiledInstruction,
     LaserstreamConfig, LaserstreamError,
 };
 use tokio::{
@@ -22,41 +20,13 @@ use tokio::{
 };
 
 use crate::channels::DlpSyncChannelsInit;
+use crate::consts::{
+    DELEGATION_PROGRAM, DELEGATION_RECORD_SIZE, MAX_PENDING_REQUESTS, MAX_PENDING_UPDATES,
+    MAX_RECONNECT_ATTEMPTS, PUBKEY_LEN,
+};
+use crate::transaction_syncer;
 use crate::types::{AccountUpdate, DlpSyncError, Pubkey, Slot};
 
-/// Size of a Solana public key in bytes.
-const PUBKEY_LEN: usize = 32;
-
-/// Delegation program address.
-const DELEGATION_PROGRAM: &str = "DELeGGvXpWV2fqJUhqcF5ZSYMS4JTLjteaAMARRSaeSh";
-
-/// Delegation program pubkey in bytes.
-const DELEGATION_PROGRAM_PUBKEY: &Pubkey = &[
-    181, 183, 0, 225, 242, 87, 58, 192, 204, 6, 34, 1, 52, 74, 207, 151, 184, 53, 6, 235, 140, 229,
-    25, 152, 204, 98, 126, 24, 147, 128, 167, 62,
-];
-
-/// Size of a delegation record account in bytes.
-const DELEGATION_RECORD_SIZE: u64 = 96;
-
-/// Instruction discriminator for undelegate operations.
-const UNDELEGATE_DISCRIMINATOR: u8 = 3;
-
-/// Length of an instruction discriminator (Anchor programs).
-const DISCRIMINATOR_LEN: usize = 8;
-
-/// Index of the delegation record account in undelegate instruction accounts.
-const DELEGATION_RECORD_ACCOUNT_INDEX: usize = 6;
-
-/// Maximum pending subscription/unsubscription requests.
-const MAX_PENDING_REQUESTS: usize = 256;
-
-/// Maximum pending account/transaction updates.
-const MAX_PENDING_UPDATES: usize = 8192;
-
-/// Maximum reconnection attempts to the Laserstream.
-const MAX_RECONNECT_ATTEMPTS: u32 = 16;
-
 /// Stream type alias for Laserstream updates.
 type LaserStream =
     Pin<Box<dyn futures::Stream<Item = Result<SubscribeUpdate, LaserstreamError>> + Send>>;
@@ -217,38 +187,9 @@ impl DlpSyncer {
 
     /// Handles a transaction update, extracting undelegations.
     fn handle_transaction_update(&self, txn: SubscribeUpdateTransaction) {
-        let Some(message) = txn
-            .transaction
-            .and_then(|t| t.transaction.zip(t.meta))
-            .and_then(|(t, m)| m.err.is_none().then_some(t.message))
-            .flatten()
-        else {
-            return;
-        };
-
-        let accounts = &message.account_keys;
-
-        let is_undelegate = |ix: &CompiledInstruction| {
-            let program_id = accounts.get(ix.program_id_index as usize)?;
-            (program_id == DELEGATION_PROGRAM_PUBKEY).then_some(())?;
-
-            let (discriminator, _) = ix.data.split_at_checked(DISCRIMINATOR_LEN)?;
-            (discriminator[0] == UNDELEGATE_DISCRIMINATOR).then_some(())?;
-
-            ix.accounts
-                .get(DELEGATION_RECORD_ACCOUNT_INDEX)
-                .and_then(|&idx| accounts.get(idx as usize))
-        };
-
-        for record_bytes in message.instructions.iter().filter_map(is_undelegate) {
-            let Ok(record) = Pubkey::try_from(record_bytes.as_slice()) else {
-                continue;
-            };
-
-            let update = AccountUpdate::Undelegated {
-                record,
-                slot: txn.slot,
-            };
+        let slot = txn.slot;
+        for record in transaction_syncer::process_update(&txn) {
+            let update = AccountUpdate::Undelegated { record, slot };
 
             if let Err(error) = self.updates.try_send(update) {
                 tracing::error!(%error, "failed to send undelegation update");
@@ -265,7 +206,6 @@ impl DlpSyncer {
     async fn connect(config: LaserstreamConfig) -> Result<LaserStream, DlpSyncError> {
         let mut accounts = HashMap::new();
         let mut slots = HashMap::new();
-        let mut transactions = HashMap::new();
 
         // Subscribe to delegation record accounts
         let account_filter = SubscribeRequestFilterAccounts {
@@ -278,11 +218,7 @@ impl DlpSyncer {
         accounts.insert("delegations".into(), account_filter);
 
         // Subscribe to undelegation transactions
-        let tx_filter = SubscribeRequestFilterTransactions {
-            account_include: vec![DELEGATION_PROGRAM.into()],
-            ..Default::default()
-        };
-        transactions.insert("undelegations".into(), tx_filter);
+        let transactions = transaction_syncer::create_filter();
 
         // Subscribe to all slot updates
         slots.insert("slots".into(), Default::default());