futu_cache/trd_cache/
backend_merge.rs

1use super::{CachedOrder, TrdCache};
2use futu_domain_trade_account::{
3    BackendOrderMergeOrderFacts, plan_backend_authoritative_order_merge_like_cpp,
4};
5use futu_domain_trade_order::{
6    OrderSnapshotReplacementPlan, OrderSnapshotRowFacts, TradeOrderFacts,
7    plan_order_snapshot_replacement_like_cpp,
8};
9
10impl TrdCache {
11    /// v1.4.90 cache saga fix: merge backend authoritative orders while retaining fresh local stubs.
12    pub fn merge_preserving_stubs(&self, acc_id: u64, backend_orders: Vec<CachedOrder>) {
13        let _ = self.replace_order_snapshot_like_cpp(acc_id, &backend_orders);
14    }
15
16    /// Injectable-clock variant used by regression tests for fresh/stale stub boundaries.
17    pub fn merge_preserving_stubs_with_now(
18        &self,
19        acc_id: u64,
20        backend_orders: Vec<CachedOrder>,
21        now_ms: u64,
22    ) {
23        let _ = self.replace_order_snapshot_like_cpp_with_now(acc_id, &backend_orders, now_ms);
24    }
25
26    /// Atomically plan C++ `UpdateOrderList` notifications and replace the
27    /// backend-authoritative snapshot while retaining Rust local compatibility
28    /// rows (fresh optimistic stubs and C++ local Deleted rows).
29    pub fn replace_order_snapshot_like_cpp(
30        &self,
31        acc_id: u64,
32        backend_orders: &[CachedOrder],
33    ) -> OrderSnapshotReplacementPlan {
34        self.replace_order_snapshot_like_cpp_with_now(acc_id, backend_orders, Self::now_ms())
35    }
36
37    /// Injectable-clock variant used by local-stub boundary tests.
38    pub fn replace_order_snapshot_like_cpp_with_now(
39        &self,
40        acc_id: u64,
41        backend_orders: &[CachedOrder],
42        now_ms: u64,
43    ) -> OrderSnapshotReplacementPlan {
44        let mut entry = self.orders.entry(acc_id).or_default();
45
46        let existing_snapshot_facts = entry
47            .iter()
48            .map(Self::order_snapshot_facts)
49            .collect::<Vec<_>>();
50        let backend_snapshot_facts = backend_orders
51            .iter()
52            .map(Self::order_snapshot_facts)
53            .collect::<Vec<_>>();
54        let replacement_plan = plan_order_snapshot_replacement_like_cpp(
55            &existing_snapshot_facts,
56            &backend_snapshot_facts,
57        );
58
59        let merge_plan = {
60            let existing_facts: Vec<_> = entry.iter().map(Self::order_merge_facts).collect();
61            let backend_facts: Vec<_> =
62                backend_orders.iter().map(Self::order_merge_facts).collect();
63            plan_backend_authoritative_order_merge_like_cpp(&existing_facts, &backend_facts, now_ms)
64        };
65
66        let mut new_orders: Vec<CachedOrder> = backend_orders
67            .iter()
68            .cloned()
69            .zip(merge_plan.backend_patches)
70            .map(|(mut order, patch)| {
71                order.is_stub = patch.is_stub;
72                order.is_local_order = patch.is_local_order;
73                order.stub_inserted_at_ms = patch.stub_inserted_at_ms;
74                order.is_pending_broker_confirm = patch.is_pending_broker_confirm;
75                order
76            })
77            .collect();
78
79        for index in merge_plan.retained_existing_indices {
80            if let Some(order) = entry.get(index) {
81                new_orders.push(order.clone());
82            }
83        }
84
85        *entry = new_orders;
86        drop(entry);
87        self.prune_order_brokers_for_acc(acc_id);
88        replacement_plan
89    }
90
91    fn order_snapshot_facts(order: &CachedOrder) -> OrderSnapshotRowFacts {
92        OrderSnapshotRowFacts {
93            order: TradeOrderFacts {
94                order_id: order.order_id,
95                backend_order_id: order.backend_order_id.clone(),
96                order_id_ex: order.order_id_ex.clone(),
97                order_status: order.order_status,
98                qty: order.qty,
99                price: order.price,
100                fill_qty: order.fill_qty,
101                aux_price: order.aux_price,
102                trail_type: order.trail_type,
103                trail_value: order.trail_value,
104                trail_spread: order.trail_spread,
105            },
106            order_version: order.order_version,
107        }
108    }
109
110    fn order_merge_facts(order: &CachedOrder) -> BackendOrderMergeOrderFacts<'_> {
111        BackendOrderMergeOrderFacts {
112            order_id: order.order_id,
113            backend_order_id: &order.backend_order_id,
114            order_id_ex: &order.order_id_ex,
115            is_stub: order.is_stub,
116            is_local_order: order.is_local_order,
117            is_local_multileg_order: order.is_local_order
118                && order.security_type
119                    == futu_proto_internal::odr_sys_cmn::SecurityType::Multileg as i32,
120            order_status: order.order_status,
121            stub_inserted_at_ms: order.stub_inserted_at_ms,
122            is_pending_broker_confirm: order.is_pending_broker_confirm,
123        }
124    }
125
126    fn now_ms() -> u64 {
127        use std::time::{SystemTime, UNIX_EPOCH};
128        SystemTime::now()
129            .duration_since(UNIX_EPOCH)
130            .map(|d| d.as_millis() as u64)
131            .unwrap_or(0)
132    }
133}
futu_cache/trd_cache/backend_merge.rs

futu_cache/trd_cache/
backend_merge.rs
