patch cypher query

src/analytics/enrich_account_funding.rs

@@ -233,7 +233,7 @@ pub fn generate_cypher_query(map: String) -> String {
         r#"
             UNWIND {map} AS account
             MERGE (sa:SwapAccount {{swap_id: account.swap_id}})
-            MERGE (ul:UserLedger {{date: datetime(account.date)}})
+            MERGE (ul:UserLedger {{swap_id: account.swap_id, date: datetime(account.date)}})
             SET ul.current_balance = account.current_balance,
                 ul.total_funded = account.total_funded,
                 ul.total_inflows = account.total_inflows,

tests/test_analytics.rs

@@ -3,7 +3,11 @@ use anyhow::Result;
 use std::path::PathBuf;
 
 use libra_forensic_db::{
-    analytics, extract_exchange_orders, load_exchange_orders,
+    analytics::{
+        self,
+        enrich_account_funding::{parse_date, BalanceTracker},
+    },
+    extract_exchange_orders, load_exchange_orders,
     neo4j_init::{get_neo4j_localhost_pool, maybe_create_indexes},
 };
 use support::neo4j_testcontainer::start_neo4j_container;
@@ -96,3 +100,139 @@ async fn test_rms_batch() -> Result<()> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn test_submit_exchange_ledger() -> Result<()> {
+    libra_forensic_db::log_setup();
+
+    let c = start_neo4j_container();
+    let port = c.get_host_port_ipv4(7687);
+    let graph = get_neo4j_localhost_pool(port).await?;
+    maybe_create_indexes(&graph).await?;
+
+    let path = env!("CARGO_MANIFEST_DIR");
+    let buf = PathBuf::from(path).join("tests/fixtures/savedOlOrders2.json");
+    let mut orders = extract_exchange_orders::read_orders_from_file(buf).unwrap();
+    assert!(orders.len() == 25450);
+
+    orders.retain(|el| {
+        if el.filled_at < parse_date("2024-01-16") {
+            if el.user == 123 {
+                return true;
+            };
+            if el.accepter == 123 {
+                return true;
+            };
+        }
+        false
+    });
+
+    assert!(orders.len() == 68);
+
+    let mut tracker = BalanceTracker::new();
+    tracker.replay_transactions(&mut orders)?;
+    dbg!(&tracker.0.len());
+    let days_records = tracker.0.len();
+    assert!(days_records == 47);
+
+    let user = tracker.0.get(&123).unwrap();
+    assert!(user.0.len() == 68);
+
+    let res = tracker.submit_one_id(123, &graph).await?;
+
+    // the number of transactions merged should equal the number of orders
+    assert!(res == orders.len() as u64);
+
+    // check there are transaction records with function args.
+    let cypher_query = neo4rs::query(
+        "MATCH (s:SwapAccount)-[r:DailyLedger]->(ul:UserLedger)
+        WHERE s.swap_id = 123
+        ORDER BY ul.date
+        RETURN s.swap_id AS id, ul.date AS date, ul.total_funded AS funded
+        ",
+    );
+
+    // Execute the query
+    let mut result = graph.execute(cypher_query).await?;
+
+    let mut prev_funding = 0;
+    let mut i = 0;
+
+    // Fetch the first row only
+    while let Some(r) = result.next().await? {
+        if let Ok(s) = r.get::<u64>("funded") {
+            i += 1;
+            assert!(s >= prev_funding, "funded totals should always increase");
+            prev_funding = s;
+        }
+    }
+
+    assert!(i == orders.len());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_submit_exchange_ledger_all() -> Result<()> {
+    libra_forensic_db::log_setup();
+
+    let c = start_neo4j_container();
+    let port = c.get_host_port_ipv4(7687);
+    let graph = get_neo4j_localhost_pool(port).await?;
+    maybe_create_indexes(&graph).await?;
+
+    let path = env!("CARGO_MANIFEST_DIR");
+    let buf = PathBuf::from(path).join("tests/fixtures/savedOlOrders2.json");
+    let mut orders = extract_exchange_orders::read_orders_from_file(buf).unwrap();
+    assert!(orders.len() == 25450);
+
+    orders.retain(|el| el.filled_at < parse_date("2024-01-16"));
+
+    dbg!(&orders.len());
+    // assert!(orders.len() == 68);
+
+    let mut tracker = BalanceTracker::new();
+    tracker.replay_transactions(&mut orders)?;
+    dbg!(&tracker.0.len());
+    let days_records = tracker.0.len();
+    // assert!(days_records == 47);
+
+    let user = tracker.0.get(&123).unwrap();
+    // assert!(user.0.len() == 68);
+
+    let res = tracker.submit_ledger(&graph).await?;
+
+    // the number of transactions merged should equal the number of orders
+    // assert!(res == orders.len() as u64);
+
+    // check there are transaction records with function args.
+    let cypher_query = neo4rs::query(
+        "MATCH (s:SwapAccount)-[r:DailyLedger]->(ul:UserLedger)
+        WHERE s.swap_id = 123
+        ORDER BY ul.date
+        RETURN s.swap_id AS id, ul.date AS date, ul.total_funded AS funded
+        ",
+    );
+
+    // Execute the query
+    let mut result = graph.execute(cypher_query).await?;
+
+    let mut prev_funding = 0;
+    let mut i = 0;
+
+    // Fetch the first row only
+    while let Some(r) = result.next().await? {
+        if let Ok(s) = r.get::<u64>("funded") {
+            i += 1;
+            dbg!(&prev_funding);
+
+            dbg!(&s);
+            assert!(s >= prev_funding, "funded totals should always increase");
+            prev_funding = s;
+        }
+    }
+
+    dbg!(&i);
+
+    Ok(())
+}