refactor: backfilling uses a transactions table

src/backfill-candles/main.rs

@@ -53,7 +53,7 @@ async fn main() -> anyhow::Result<()> {
 
 async fn save_candles(candles: Vec<Candle>, client: Object) -> anyhow::Result<()> {
     if !candles.is_empty() {
-        let upsert_statement = build_candles_upsert_statement(candles);
+        let upsert_statement = build_candles_upsert_statement(&candles);
         client
             .execute(&upsert_statement, &[])
             .await

src/backfill-trades/main.rs

@@ -1,23 +1,28 @@
 use anchor_lang::prelude::Pubkey;
 use chrono::{DateTime, Duration, NaiveDateTime, Utc};
+use deadpool_postgres::Pool;
 use futures::future::join_all;
+use log::debug;
 use openbook_candles::{
-    database::{initialize::connect_to_database, insert::persist_fill_events},
+    database::{
+        fetch::fetch_worker_transactions,
+        initialize::{connect_to_database, setup_database},
+        insert::{add_fills_atomically, build_transactions_insert_statement},
+    },
     structs::{
         markets::{fetch_market_infos, load_markets},
-        openbook::OpenBookFillEvent,
+        transaction::{PgTransaction, NUM_TRANSACTION_PARTITIONS},
     },
-    utils::Config,
+    utils::{AnyhowWrap, Config, OPENBOOK_KEY},
     worker::trade_fetching::parsing::parse_trades_from_openbook_txns,
 };
 use solana_client::{
     nonblocking::rpc_client::RpcClient, rpc_client::GetConfirmedSignaturesForAddress2Config,
-    rpc_config::RpcTransactionConfig, rpc_response::RpcConfirmedTransactionStatusWithSignature,
+    rpc_config::RpcTransactionConfig
 };
 use solana_sdk::{commitment_config::CommitmentConfig, signature::Signature};
 use solana_transaction_status::UiTransactionEncoding;
-use std::{collections::HashMap, env, str::FromStr};
+use std::{collections::HashMap, env, str::FromStr,time::Duration as WaitDuration };
-use tokio::sync::mpsc::{self, Sender};
 
 #[tokio::main(flavor = "multi_thread", worker_threads = 10)]
 async fn main() -> anyhow::Result<()> {
@@ -26,6 +31,8 @@ async fn main() -> anyhow::Result<()> {
     assert!(args.len() == 2);
 
     let path_to_markets_json = &args[1];
+    // let num_days = args[2].parse::<i64>().unwrap(); // TODO: implement
+    let num_days = 1;
     let rpc_url: String = dotenv::var("RPC_URL").unwrap();
 
     let config = Config {
@@ -40,147 +47,147 @@ async fn main() -> anyhow::Result<()> {
     println!("{:?}", target_markets);
 
     let pool = connect_to_database().await?;
-    let (fill_sender, mut fill_receiver) = mpsc::channel::<OpenBookFillEvent>(1000);
+    setup_database(&pool).await?;
 
-    tokio::spawn(async move {
+    let mut handles = vec![];
-        loop {
+
-            persist_fill_events(&pool, &mut fill_receiver)
+    let rpc_clone = rpc_url.clone();
+    let pool_clone = pool.clone();
+    handles.push(tokio::spawn(async move {
+        fetch_signatures(rpc_clone, &pool_clone, num_days).await.unwrap();
+    }));
+
+    for id in 0..NUM_TRANSACTION_PARTITIONS {
+        let rpc_clone = rpc_url.clone();
+        let pool_clone = pool.clone();
+        let markets_clone = target_markets.clone();
+        handles.push(tokio::spawn(async move {
+            backfill(id as i32, rpc_clone, &pool_clone, &markets_clone)
                 .await
                 .unwrap();
-        }
+        }));
-    });
+    }
 
-    backfill(rpc_url, &fill_sender, &target_markets).await?;
+    // TODO: spawn status thread
+
+    futures::future::join_all(handles).await;
+    Ok(())
+}
+
+pub async fn fetch_signatures(rpc_url: String, pool: &Pool, num_days: i64) -> anyhow::Result<()> {
+    let mut before_sig: Option<Signature> = None;
+    let mut now_time = Utc::now().timestamp();
+    let end_time = (Utc::now() - Duration::days(num_days)).timestamp();
+    let rpc_client = RpcClient::new_with_commitment(rpc_url.clone(), CommitmentConfig::confirmed());
+
+    while now_time > end_time {
+        let rpc_config = GetConfirmedSignaturesForAddress2Config {
+            before: before_sig,
+            until: None,
+            limit: None,
+            commitment: Some(CommitmentConfig::confirmed()),
+        };
+
+        let sigs = match rpc_client
+            .get_signatures_for_address_with_config(
+                &OPENBOOK_KEY,
+                rpc_config,
+            )
+            .await
+        {
+            Ok(sigs) => sigs,
+            Err(e) => {
+                println!("Error fetching signatures: {}", e);
+                continue;
+            }
+        };
+        if sigs.is_empty() {
+            println!("No signatures found, trying again");
+            continue;
+        }
+        let last = sigs.last().unwrap();
+        let last_time = last.block_time.unwrap().clone();
+        let last_signature = last.signature.clone();
+        let transactions = sigs
+            .into_iter()
+            .map(|s| PgTransaction::from_rpc_confirmed_transaction(s))
+            .collect::<Vec<PgTransaction>>();
+
+        if transactions.is_empty() {
+            println!("No transactions found, trying again");   
+        }
+        debug!("writing: {:?} txns to DB\n", transactions.len());
+        let upsert_statement = build_transactions_insert_statement(transactions);
+            let client = pool.get().await?;
+            client
+                .execute(&upsert_statement, &[])
+                .await
+                .map_err_anyhow()?;
+
+        now_time = last_time;
+        before_sig = Some(Signature::from_str(&last_signature)?);
+        let time_left = backfill_time_left(now_time, end_time);
+        println!(
+            "{} minutes ~ {} days remaining in the backfill\n",
+            time_left.num_minutes(),
+            time_left.num_days()
+        );
+    }
     Ok(())
 }
 
 pub async fn backfill(
+    worker_id: i32,
     rpc_url: String,
-    fill_sender: &Sender<OpenBookFillEvent>,
+    pool: &Pool,
     target_markets: &HashMap<Pubkey, String>,
 ) -> anyhow::Result<()> {
-    println!("backfill started");
+    println!("Worker {} up \n", worker_id);
-    let mut before_sig: Option<Signature> = None;
+    let rpc_client = RpcClient::new_with_commitment(rpc_url, CommitmentConfig::confirmed());
-    let mut now_time = Utc::now().timestamp();
-    let end_time = (Utc::now() - Duration::days(1)).timestamp();
 
-    let mut handles = vec![];
+    loop {
+        let transactions = fetch_worker_transactions(worker_id, pool).await?;
+        if transactions.len() == 0 {
+            println!("No signatures found by worker {}", worker_id);
+            tokio::time::sleep(WaitDuration::from_secs(1)).await;
+            continue;
+        };
 
-    while now_time > end_time {
+        // for each signature, fetch the transaction
-        let rpc_client =
+        let txn_config = RpcTransactionConfig {
-            RpcClient::new_with_commitment(rpc_url.clone(), CommitmentConfig::confirmed());
+            encoding: Some(UiTransactionEncoding::Json),
-        let maybe_r = get_signatures(&rpc_client, before_sig).await;
+            commitment: Some(CommitmentConfig::confirmed()),
+            max_supported_transaction_version: Some(0),
+        };
 
-        match maybe_r {
+        let sig_strings = transactions
-            Some((last, time, sigs)) => {
+            .iter()
-                now_time = time;
+            .map(|t| t.signature.clone())
-                before_sig = Some(last);
+            .collect::<Vec<String>>();
-                let time_left = backfill_time_left(now_time, end_time);
-                println!(
-                    "{} minutes ~ {} days remaining in the backfill\n",
-                    time_left.num_minutes(),
-                    time_left.num_days()
-                );
 
-                let cloned_markets = target_markets.clone();
+        let signatures: Vec<_> = transactions
-                let cloned_sender = fill_sender.clone();
+            .into_iter()
-                let handle = tokio::spawn(async move {
+            .map(|t| t.signature.parse::<Signature>().unwrap())
-                    get_transactions(&rpc_client, sigs, &cloned_sender, &cloned_markets).await;
+            .collect();
-                });
+
-                handles.push(handle);
+        let txn_futs: Vec<_> = signatures
-            }
+            .iter()
-            None => {}
+            .map(|s| rpc_client.get_transaction_with_config(s, txn_config))
-        }
+            .collect();
+
+        let mut txns = join_all(txn_futs).await;
+
+        let fills = parse_trades_from_openbook_txns(&mut txns, target_markets);
+
+        // Write any fills to the database, and mark the transactions as processed
+        add_fills_atomically(pool, worker_id, fills, sig_strings).await?;
     }
 
-    futures::future::join_all(handles).await;
+    // TODO: graceful shutdown
-
+    // println!("Worker {} down \n", worker_id);
-    println!("Backfill complete \n");
     Ok(())
 }
 
-pub async fn get_signatures(
-    rpc_client: &RpcClient,
-    before_sig: Option<Signature>,
-) -> Option<(
-    Signature,
-    i64,
-    Vec<RpcConfirmedTransactionStatusWithSignature>,
-)> {
-    let rpc_config = GetConfirmedSignaturesForAddress2Config {
-        before: before_sig,
-        until: None,
-        limit: None,
-        commitment: Some(CommitmentConfig::confirmed()),
-    };
-
-    let sigs = match rpc_client
-        .get_signatures_for_address_with_config(
-            &Pubkey::from_str("srmqPvymJeFKQ4zGQed1GFppgkRHL9kaELCbyksJtPX").unwrap(),
-            rpc_config,
-        )
-        .await
-    {
-        Ok(s) => s,
-        Err(e) => {
-            println!("Error in get_signatures_for_address_with_config: {}", e);
-            return None;
-        }
-    };
-
-    if sigs.is_empty() {
-        println!("No signatures found");
-        return None;
-    }
-    let last = sigs.last().unwrap();
-    // println!("{:?}", last.block_time.unwrap());
-    Some((
-        Signature::from_str(&last.signature).unwrap(),
-        last.block_time.unwrap(),
-        sigs,
-    ))
-}
-
-pub async fn get_transactions(
-    rpc_client: &RpcClient,
-    mut sigs: Vec<RpcConfirmedTransactionStatusWithSignature>,
-    fill_sender: &Sender<OpenBookFillEvent>,
-    target_markets: &HashMap<Pubkey, String>,
-) {
-    sigs.retain(|sig| sig.err.is_none());
-    if sigs.last().is_none() {
-        return;
-    }
-
-    let txn_config = RpcTransactionConfig {
-        encoding: Some(UiTransactionEncoding::Json),
-        commitment: Some(CommitmentConfig::confirmed()),
-        max_supported_transaction_version: Some(0),
-    };
-
-    let signatures: Vec<_> = sigs
-        .into_iter()
-        .map(|sig| sig.signature.parse::<Signature>().unwrap())
-        .collect();
-
-    let txn_futs: Vec<_> = signatures
-        .iter()
-        .map(|s| rpc_client.get_transaction_with_config(s, txn_config))
-        .collect();
-
-    let mut txns = join_all(txn_futs).await;
-
-    let fills = parse_trades_from_openbook_txns(&mut txns, target_markets);
-    if !fills.is_empty() {
-        for fill in fills.into_iter() {
-            // println!("Sending fill {:?}", fill);
-            if let Err(_) = fill_sender.send(fill).await {
-                panic!("receiver dropped");
-            }
-        }
-    }
-}
-
 fn backfill_time_left(current_time: i64, backfill_end: i64) -> Duration {
     let naive_cur = NaiveDateTime::from_timestamp_millis(current_time * 1000).unwrap();
     let naive_bf = NaiveDateTime::from_timestamp_millis(backfill_end * 1000).unwrap();

src/database/fetch.rs

@@ -4,6 +4,7 @@ use crate::structs::{
     openbook::PgOpenBookFill,
     resolution::Resolution,
     trader::PgTrader,
+    transaction::PgTransaction,
 };
 use chrono::{DateTime, Utc};
 use deadpool_postgres::{GenericClient, Pool};
@@ -320,3 +321,23 @@ pub async fn fetch_coingecko_24h_high_low(
         .map(PgCoinGecko24HighLow::from_row)
         .collect())
 }
+
+/// Fetches unprocessed, non-error transactions for the specified worker partition.
+/// Pulls at most 50 transactions at a time.
+pub async fn fetch_worker_transactions(
+    worker_id: i32,
+    pool: &Pool,
+) -> anyhow::Result<Vec<PgTransaction>> {
+    let client = pool.get().await?;
+
+    let stmt = r#"SELECT signature, program_pk, block_datetime, slot, err, "processed", worker_partition
+            FROM transactions
+            where worker_partition = $1
+            and err = false 
+            and processed = false
+            LIMIT 50"#;
+
+    let rows = client.query(stmt, &[&worker_id]).await?;
+
+    Ok(rows.into_iter().map(PgTransaction::from_row).collect())
+}

src/database/initialize.rs

@@ -67,8 +67,9 @@ pub async fn connect_to_database() -> anyhow::Result<Pool> {
 
 pub async fn setup_database(pool: &Pool) -> anyhow::Result<()> {
     let candles_table_fut = create_candles_table(pool);
+    let transactions_table_fut = create_transactions_table(pool);
     let fills_table_fut = create_fills_table(pool);
-    let result = tokio::try_join!(candles_table_fut, fills_table_fut);
+    let result = tokio::try_join!(candles_table_fut, transactions_table_fut, fills_table_fut);
     match result {
         Ok(_) => {
             println!("Successfully configured database");
@@ -137,7 +138,8 @@ pub async fn create_fills_table(pool: &Pool) -> anyhow::Result<()> {
             native_qty_received double precision not null,
             native_fee_or_rebate double precision not null,
             fee_tier text not null,
-            order_id text not null
+            order_id text not null,
+            log_index int4 not null
         )",
             &[],
         )
@@ -158,3 +160,41 @@ pub async fn create_fills_table(pool: &Pool) -> anyhow::Result<()> {
         .await?;
     Ok(())
 }
+
+pub async fn create_transactions_table(pool: &Pool) -> anyhow::Result<()> {
+    let client = pool.get().await?;
+
+    client
+        .execute(
+            "CREATE TABLE IF NOT EXISTS transactions (
+                signature text NOT NULL,
+                program_pk text NOT NULL,
+                block_datetime timestamptz NOT NULL,
+                slot bigint NOT NULL,
+                err bool NOT NULL,
+                processed bool NOT NULL,
+                worker_partition int4 NOT NULL,
+                CONSTRAINT transactions_pk PRIMARY KEY (signature, worker_partition)
+            ) PARTITION BY LIST (worker_partition);",
+            &[],
+        )
+        .await?;
+
+    client.batch_execute(
+        "CREATE INDEX IF NOT EXISTS transactions_processed_err_idx ON ONLY transactions (signature) WHERE processed IS NOT TRUE and err IS NOT TRUE;
+        CREATE INDEX IF NOT EXISTS transactions_program_pk_idx ON ONLY transactions USING btree (program_pk, slot DESC);
+
+        CREATE TABLE IF NOT EXISTS transactions_0 PARTITION OF transactions  FOR VALUES IN (0);
+        CREATE TABLE IF NOT EXISTS transactions_1 PARTITION OF transactions  FOR VALUES IN (1);
+        CREATE TABLE IF NOT EXISTS transactions_2 PARTITION OF transactions  FOR VALUES IN (2);
+        CREATE TABLE IF NOT EXISTS transactions_3 PARTITION OF transactions  FOR VALUES IN (3);
+        CREATE TABLE IF NOT EXISTS transactions_4 PARTITION OF transactions  FOR VALUES IN (4);
+        CREATE TABLE IF NOT EXISTS transactions_5 PARTITION OF transactions  FOR VALUES IN (5);
+        CREATE TABLE IF NOT EXISTS transactions_6 PARTITION OF transactions  FOR VALUES IN (6);
+        CREATE TABLE IF NOT EXISTS transactions_7 PARTITION OF transactions  FOR VALUES IN (7);
+        CREATE TABLE IF NOT EXISTS transactions_8 PARTITION OF transactions  FOR VALUES IN (8);
+        CREATE TABLE IF NOT EXISTS transactions_9 PARTITION OF transactions  FOR VALUES IN (9);"
+    ).await?;
+
+    Ok(())
+}

src/database/insert.rs

@@ -7,10 +7,44 @@ use std::{
 use tokio::sync::mpsc::{error::TryRecvError, Receiver};
 
 use crate::{
-    structs::{candle::Candle, openbook::OpenBookFillEvent},
+    structs::{candle::Candle, openbook::OpenBookFillEvent, transaction::PgTransaction},
     utils::{to_timestampz, AnyhowWrap},
 };
 
+pub async fn add_fills_atomically(
+    pool: &Pool,
+    worker_id: i32,
+    fills: Vec<OpenBookFillEvent>,
+    signatures: Vec<String>,
+) -> anyhow::Result<()> {
+    let mut client = pool.get().await?;
+
+    let db_txn = client.build_transaction().start().await?;
+
+    // 1. Insert fills
+    if fills.len() > 0 {
+        let fills_statement = build_fills_upsert_statement_not_crazy(fills);
+        db_txn
+            .execute(&fills_statement, &[])
+            .await
+            .map_err_anyhow()
+            .unwrap();
+    }
+
+    // 2. Update txns table as processed
+    let transactions_statement =
+        build_transactions_processed_update_statement(worker_id, signatures);
+    db_txn
+        .execute(&transactions_statement, &[])
+        .await
+        .map_err_anyhow()
+        .unwrap();
+
+    db_txn.commit().await?;
+
+    Ok(())
+}
+
 pub async fn persist_fill_events(
     pool: &Pool,
     fill_receiver: &mut Receiver<OpenBookFillEvent>,
@@ -50,12 +84,12 @@ pub async fn persist_fill_events(
 
 #[allow(deprecated)]
 fn build_fills_upsert_statement(events: HashMap<OpenBookFillEvent, u8>) -> String {
-    let mut stmt = String::from("INSERT INTO fills (id, time, market, open_orders, open_orders_owner, bid, maker, native_qty_paid, native_qty_received, native_fee_or_rebate, fee_tier, order_id) VALUES");
+    let mut stmt = String::from("INSERT INTO fills (id, time, market, open_orders, open_orders_owner, bid, maker, native_qty_paid, native_qty_received, native_fee_or_rebate, fee_tier, order_id, log_index) VALUES");
     for (idx, event) in events.keys().enumerate() {
         let mut hasher = DefaultHasher::new();
         event.hash(&mut hasher);
         let val_str = format!(
-            "({}, \'{}\', \'{}\', \'{}\', \'{}\', {}, {}, {}, {}, {}, {}, {})",
+            "({}, \'{}\', \'{}\', \'{}\', \'{}\', {}, {}, {}, {}, {}, {}, {}, {})",
             hasher.finish(),
             to_timestampz(event.block_time as u64).to_rfc3339(),
             event.market,
@@ -68,6 +102,7 @@ fn build_fills_upsert_statement(events: HashMap<OpenBookFillEvent, u8>) -> Strin
             event.native_fee_or_rebate,
             event.fee_tier,
             event.order_id,
+            event.log_index,
         );
 
         if idx == 0 {
@@ -77,7 +112,42 @@ fn build_fills_upsert_statement(events: HashMap<OpenBookFillEvent, u8>) -> Strin
         }
     }
 
-    let handle_conflict = "ON CONFLICT (id) DO UPDATE SET market=excluded.market";
+    let handle_conflict = "ON CONFLICT DO NOTHING";
+
+    stmt = format!("{} {}", stmt, handle_conflict);
+    stmt
+}
+
+fn build_fills_upsert_statement_not_crazy(fills: Vec<OpenBookFillEvent>) -> String {
+    let mut stmt = String::from("INSERT INTO fills (id, time, market, open_orders, open_orders_owner, bid, maker, native_qty_paid, native_qty_received, native_fee_or_rebate, fee_tier, order_id, log_index) VALUES");
+    for (idx, fill) in fills.iter().enumerate() {
+        let mut hasher = DefaultHasher::new();
+        fill.hash(&mut hasher);
+        let val_str = format!(
+            "({}, \'{}\', \'{}\', \'{}\', \'{}\', {}, {}, {}, {}, {}, {}, {}, {})",
+            hasher.finish(),
+            to_timestampz(fill.block_time as u64).to_rfc3339(),
+            fill.market,
+            fill.open_orders,
+            fill.open_orders_owner,
+            fill.bid,
+            fill.maker,
+            fill.native_qty_paid,
+            fill.native_qty_received,
+            fill.native_fee_or_rebate,
+            fill.fee_tier,
+            fill.order_id,
+            fill.log_index,
+        );
+
+        if idx == 0 {
+            stmt = format!("{} {}", &stmt, val_str);
+        } else {
+            stmt = format!("{}, {}", &stmt, val_str);
+        }
+    }
+
+    let handle_conflict = "ON CONFLICT DO NOTHING";
 
     stmt = format!("{} {}", stmt, handle_conflict);
     stmt
@@ -121,6 +191,57 @@ pub fn build_candles_upsert_statement(candles: &Vec<Candle>) -> String {
     stmt
 }
 
+pub fn build_transactions_insert_statement(transactions: Vec<PgTransaction>) -> String {
+    let mut stmt = String::from("INSERT INTO transactions (signature, program_pk, block_datetime, slot, err, processed, worker_partition) VALUES");
+    for (idx, txn) in transactions.iter().enumerate() {
+        let val_str = format!(
+            "(\'{}\', \'{}\', \'{}\', \'{}\', {}, {}, {})",
+            txn.signature,
+            txn.program_pk,
+            txn.block_datetime.to_rfc3339(),
+            txn.slot,
+            txn.err,
+            txn.processed,
+            txn.worker_partition,
+        );
+
+        if idx == 0 {
+            stmt = format!("{} {}", &stmt, val_str);
+        } else {
+            stmt = format!("{}, {}", &stmt, val_str);
+        }
+    }
+
+    let handle_conflict = "ON CONFLICT DO NOTHING";
+
+    stmt = format!("{} {}", stmt, handle_conflict);
+    stmt
+}
+
+pub fn build_transactions_processed_update_statement(
+    worker_id: i32,
+    processed_signatures: Vec<String>,
+) -> String {
+    let mut stmt = String::from(
+        "UPDATE transactions
+    SET processed = true
+    WHERE transactions.signature IN (",
+    );
+    for (idx, sig) in processed_signatures.iter().enumerate() {
+        let val_str = if idx == processed_signatures.len() - 1 {
+            format!("\'{}\'", sig,)
+        } else {
+            format!("\'{}\',", sig,)
+        };
+        stmt = format!("{} {}", &stmt, val_str);
+    }
+
+    let worker_stmt = format!(") AND worker_partition = {} ", worker_id);
+
+    stmt = format!("{} {}", stmt, worker_stmt);
+    stmt
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -145,6 +266,7 @@ mod tests {
             client_order_id: None,
             referrer_rebate: Some(841),
             block_time: 0,
+            log_index: 1,
         };
 
         let event_2 = OpenBookFillEvent {
@@ -163,6 +285,7 @@ mod tests {
             client_order_id: None,
             referrer_rebate: Some(841),
             block_time: 0,
+            log_index: 1,
         };
 
         let mut h1 = DefaultHasher::new();

src/structs/mod.rs

@@ -6,3 +6,4 @@ pub mod resolution;
 pub mod slab;
 pub mod trader;
 pub mod tradingview;
+pub mod transaction;

src/structs/openbook.rs

@@ -22,7 +22,7 @@ pub struct OpenBookFillEventRaw {
     pub referrer_rebate: Option<u64>,
 }
 impl OpenBookFillEventRaw {
-    pub fn with_time(self, block_time: i64) -> OpenBookFillEvent {
+    pub fn into_event(self, block_time: i64, log_index: usize) -> OpenBookFillEvent {
         OpenBookFillEvent {
             market: self.market,
             open_orders: self.open_orders,
@@ -38,6 +38,7 @@ impl OpenBookFillEventRaw {
             client_order_id: self.client_order_id,
             referrer_rebate: self.referrer_rebate,
             block_time,
+            log_index,
         }
     }
 }
@@ -59,6 +60,7 @@ pub struct OpenBookFillEvent {
     pub client_order_id: Option<u64>,
     pub referrer_rebate: Option<u64>,
     pub block_time: i64,
+    pub log_index: usize
 }
 
 #[derive(Copy, Clone, Debug, PartialEq)]

src/structs/transaction.rs

@@ -0,0 +1,52 @@
+use chrono::{DateTime, Utc};
+use solana_client::rpc_response::RpcConfirmedTransactionStatusWithSignature;
+use tokio_postgres::Row;
+
+use crate::utils::{to_timestampz, OPENBOOK_KEY};
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct PgTransaction {
+    pub signature: String,
+    pub program_pk: String,
+    pub block_datetime: DateTime<Utc>,
+    pub slot: u64,
+    pub err: bool,
+    pub processed: bool,
+    pub worker_partition: i32,
+}
+
+pub const NUM_TRANSACTION_PARTITIONS: u64 = 10;
+
+impl PgTransaction {
+    pub fn from_rpc_confirmed_transaction(
+        rpc_confirmed_transaction: RpcConfirmedTransactionStatusWithSignature,
+    ) -> Self {
+        PgTransaction {
+            signature: rpc_confirmed_transaction.signature,
+            program_pk: OPENBOOK_KEY.to_string(),
+            block_datetime: to_timestampz(rpc_confirmed_transaction.block_time.unwrap() as u64),
+            slot: rpc_confirmed_transaction.slot,
+            err: rpc_confirmed_transaction.err.is_some(),
+            processed: false,
+            worker_partition: (rpc_confirmed_transaction.slot % NUM_TRANSACTION_PARTITIONS) as i32,
+        }
+    }
+
+    pub fn from_row(row: Row) -> Self {
+        let slot_raw = row.get::<usize, i64>(3);
+        PgTransaction {
+            signature: row.get(0),
+            program_pk: row.get(1),
+            block_datetime: row.get(2),
+            slot: slot_raw as u64,
+            err: row.get(4),
+            processed: row.get(5),
+            worker_partition: row.get(6),
+        }
+    }
+}
+
+pub enum ProcessState {
+    Processed,
+    Unprocessed,
+}

src/utils/mod.rs

@@ -1,9 +1,13 @@
+use anchor_lang::prelude::Pubkey;
 use chrono::{NaiveDateTime, Utc};
 use deadpool_postgres::Pool;
 use serde_derive::Deserialize;
+use solana_sdk::pubkey;
 
 use crate::structs::markets::MarketInfo;
 
+pub const OPENBOOK_KEY: Pubkey = pubkey!("srmqPvymJeFKQ4zGQed1GFppgkRHL9kaELCbyksJtPX");
+
 pub trait AnyhowWrap {
     type Value;
     fn map_err_anyhow(self) -> anyhow::Result<Self::Value>;

src/worker/main.rs

@@ -9,7 +9,7 @@ use openbook_candles::worker::trade_fetching::scrape::scrape;
 use openbook_candles::{
     database::{
         initialize::{connect_to_database, setup_database},
-        insert::{persist_fill_events},
+        insert::persist_fill_events,
     },
     worker::candle_batching::batch_for_market,
 };

src/worker/trade_fetching/parsing.rs

@@ -55,7 +55,7 @@ fn parse_openbook_fills_from_logs(
     block_time: i64,
 ) -> Option<Vec<OpenBookFillEvent>> {
     let mut fills_vector = Vec::<OpenBookFillEvent>::new();
-    for l in logs {
+    for (idx, l) in logs.iter().enumerate() {
         match l.strip_prefix(PROGRAM_DATA) {
             Some(log) => {
                 let borsh_bytes = match anchor_lang::__private::base64::decode(log) {
@@ -68,7 +68,7 @@ fn parse_openbook_fills_from_logs(
 
                 match event {
                     Ok(e) => {
-                        let fill_event = e.with_time(block_time);
+                        let fill_event = e.into_event(block_time, idx);
                         if target_markets.contains_key(&fill_event.market) {
                             fills_vector.push(fill_event);
                         }

src/worker/trade_fetching/scrape.rs

@@ -11,7 +11,7 @@ use tokio::sync::mpsc::Sender;
 
 use crate::{
     structs::openbook::OpenBookFillEvent,
-    utils::Config,
+    utils::{Config, OPENBOOK_KEY},
     worker::metrics::{METRIC_FILLS_TOTAL, METRIC_RPC_ERRORS_TOTAL},
 };
 
@@ -55,7 +55,7 @@ pub async fn scrape_transactions(
 
     let mut sigs = match rpc_client
         .get_signatures_for_address_with_config(
-            &Pubkey::from_str("srmqPvymJeFKQ4zGQed1GFppgkRHL9kaELCbyksJtPX").unwrap(),
+            &OPENBOOK_KEY,
             rpc_config,
         )
         .await