e2e_storage_read/
e2e_storage_read.rs

1use std::path::PathBuf;
2use digdigdig3_station::{StorageManager, StorageConfig, StreamKey};
3
4#[tokio::main]
5async fn main() -> Result<(), Box<dyn std::error::Error>> {
6    let cfg = StorageConfig {
7        root: PathBuf::from("./e2e_test_data"),
8        default_retention_days: 365,
9        orderbook_snapshot_interval_secs: 0,
10    };
11    let mgr = StorageManager::new(cfg)?;
12
13    for (exchange, account, sym, stream) in [
14        ("binance", "spot", "BTCUSDT",  "trade"),
15        ("binance", "spot", "BTCUSDT",  "ticker"),
16        ("bybit",   "spot", "BTCUSDT",  "trade"),
17        ("bybit",   "spot", "BTCUSDT",  "ticker"),
18        ("okx",     "spot", "BTC-USDT", "trade"),
19        ("okx",     "spot", "BTC-USDT", "ticker"),
20    ] {
21        let key = StreamKey {
22            exchange: exchange.to_string(),
23            account: account.to_string(),
24            symbol: sym.to_string(),
25            stream_kind: stream.to_string(),
26        };
27        // read today only — avoid iterating thousands of days from epoch
28        let now_ms = chrono::Utc::now().timestamp_millis();
29        let day_start = now_ms - (now_ms % 86_400_000);
30        let records = mgr.read_range(&key, day_start, now_ms).await?;
31        let first_ts = records.first().map(|(t, _)| *t);
32        let last_ts = records.last().map(|(t, _)| *t);
33        let span_secs = match (first_ts, last_ts) {
34            (Some(f), Some(l)) => (l - f) / 1000,
35            _ => 0,
36        };
37        println!("{exchange}:{sym}:{stream:8} records={} span={span_secs}s first_ts={first_ts:?} last_ts={last_ts:?}",
38            records.len());
39
40        if let Some((_, payload)) = records.first() {
41            match serde_json::from_slice::<serde_json::Value>(payload) {
42                Ok(v) => {
43                    let keys: Vec<_> = v.as_object()
44                        .map(|o| o.keys().map(|k| k.as_str()).collect())
45                        .unwrap_or_default();
46                    println!("  first event keys: {:?}", keys);
47                }
48                Err(e) => println!("  first event PARSE FAIL: {e}"),
49            }
50        }
51    }
52    Ok(())
53}
e2e_storage_read/e2e_storage_read.rs

e2e_storage_read/
e2e_storage_read.rs
