digdigdig3-station 0.3.21

Consumer-facing builder over digdigdig3 ExchangeHub. Persistence, cache, replay, cure, orderbook tracker, multiplex, reconnect.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

//! Combined repair pipeline.
//!
//! Runs integrity check → dedup → gap detection in sequence, returning a single
//! [`RepairReport`].  Pass `dry_run = true` to skip the actual dedup write.

use crate::cure::{
    dedup::Deduper,
    gap::{GapDetector, GapInfo},
    integrity::{IntegrityChecker, IntegrityReport},
};
use crate::storage::{StorageManager, StreamKey};

// ── RepairReport ──────────────────────────────────────────────────────────────

/// Combined output of one pipeline run.
#[derive(Debug)]
pub struct RepairReport {
    pub integrity: IntegrityReport,
    /// Records kept after deduplication (or `record_count` when `dry_run = true`).
    pub deduped_kept: u64,
    /// Records removed as duplicates (or `duplicate_count` when `dry_run = true`).
    pub deduped_removed: u64,
    /// Sequence gaps found during orderbook replay (empty for non-orderbook streams).
    pub orderbook_gaps: Vec<GapInfo>,
}

// ── RepairPipeline ────────────────────────────────────────────────────────────

/// Runs integrity → dedup → gap detection on a stored stream.
pub struct RepairPipeline<'a> {
    storage: &'a StorageManager,
}

impl<'a> RepairPipeline<'a> {
    pub fn new(storage: &'a StorageManager) -> Self {
        Self { storage }
    }

    /// Run the full pipeline over `[from_ms, to_ms]`.
    ///
    /// When `dry_run = true`, deduplication is simulated: counts are taken from
    /// `IntegrityReport` and no files are written.
    ///
    /// Gap detection is only performed when `key.stream_kind` contains `"orderbook"`,
    /// `"Orderbook"`, or `"delta"` / `"Delta"` (case-insensitive check).
    pub async fn run(
        &self,
        key: &StreamKey,
        from_ms: i64,
        to_ms: i64,
        dry_run: bool,
    ) -> std::io::Result<RepairReport> {
        let integrity = IntegrityChecker::new(self.storage)
            .check(key, from_ms, to_ms)
            .await?;

        let (kept, removed) = if dry_run {
            (integrity.record_count, integrity.duplicate_count)
        } else {
            Deduper::new(self.storage)
                .dedup(key, from_ms, to_ms)
                .await?
        };

        let kind_lower = key.stream_kind.to_lowercase();
        let is_orderbook = kind_lower.contains("orderbook") || kind_lower.contains("delta");

        let orderbook_gaps = if is_orderbook {
            GapDetector::new(self.storage)
                .detect(key, from_ms, to_ms)
                .await?
        } else {
            vec![]
        };

        Ok(RepairReport {
            integrity,
            deduped_kept: kept,
            deduped_removed: removed,
            orderbook_gaps,
        })
    }
}