futu_cache/
price_reminder_state.rs

1//! Shared runtime state for CMD6803/CMD6808/CMD6809 price reminders.
2
3use std::collections::{HashMap, HashSet};
4use std::sync::Arc;
5use std::time::{Duration, Instant};
6
7use futu_domain_qot_price_reminder::{
8    PriceReminderCachedItem, PriceReminderChangePushAction, PriceReminderChangeState,
9    PriceReminderSetCacheAction,
10};
11use parking_lot::Mutex;
12
13/// C++ `ClearTimePause = 60 * 60`.
14/// Ref: `NNData_PriceReminder.cpp:4,235-269`.
15pub const PRICE_REMINDER_IDLE_TTL: Duration = Duration::from_secs(60 * 60);
16
17#[derive(Clone, Debug, Eq, PartialEq)]
18pub enum PriceReminderItemLookup {
19    RefreshRequired,
20    Missing,
21    Found(PriceReminderCachedItem),
22}
23
24#[derive(Debug, Default)]
25struct PriceReminderStateInner {
26    by_stock: HashMap<u64, HashMap<i64, PriceReminderCachedItem>>,
27    key_to_stock: HashMap<i64, u64>,
28    dirty_stocks: HashSet<u64>,
29    last_access: HashMap<u64, Instant>,
30    change_state: PriceReminderChangeState,
31}
32
33#[derive(Debug, Default)]
34pub struct PriceReminderStateCache {
35    inner: Mutex<PriceReminderStateInner>,
36}
37
38impl PriceReminderStateCache {
39    #[must_use]
40    pub fn new() -> Arc<Self> {
41        Arc::new(Self::default())
42    }
43
44    pub fn replace_stock(&self, stock_id: u64, items: Vec<PriceReminderCachedItem>) {
45        self.replace_stock_at(stock_id, items, Instant::now());
46    }
47
48    pub fn replace_stock_at(
49        &self,
50        stock_id: u64,
51        items: Vec<PriceReminderCachedItem>,
52        now: Instant,
53    ) {
54        let mut inner = self.inner.lock();
55        remove_stock_items(&mut inner, stock_id);
56
57        let mut replacement = HashMap::with_capacity(items.len());
58        for mut item in items {
59            // C++ indexes CMD6808 rows by the outer requested nSecID even when
60            // an inner row disagrees (`NNData_PriceReminder.cpp:44-55`).
61            item.stock_id = stock_id;
62            remove_key_from_previous_stock(&mut inner, item.key, stock_id);
63            inner.key_to_stock.insert(item.key, stock_id);
64            replacement.insert(item.key, item);
65        }
66        inner.by_stock.insert(stock_id, replacement);
67        inner.dirty_stocks.remove(&stock_id);
68        inner.last_access.insert(stock_id, now);
69    }
70
71    #[must_use]
72    pub fn lookup(&self, stock_id: u64, key: i64) -> PriceReminderItemLookup {
73        self.lookup_at(stock_id, key, Instant::now())
74    }
75
76    #[must_use]
77    pub fn lookup_at(&self, stock_id: u64, key: i64, now: Instant) -> PriceReminderItemLookup {
78        let mut inner = self.inner.lock();
79        inner.last_access.insert(stock_id, now);
80        if inner.dirty_stocks.contains(&stock_id) {
81            return PriceReminderItemLookup::RefreshRequired;
82        }
83        let Some(items) = inner.by_stock.get(&stock_id) else {
84            return PriceReminderItemLookup::RefreshRequired;
85        };
86        items.get(&key).cloned().map_or(
87            PriceReminderItemLookup::Missing,
88            PriceReminderItemLookup::Found,
89        )
90    }
91
92    #[must_use]
93    pub fn stock_id_for_key(&self, key: i64) -> Option<u64> {
94        self.inner.lock().key_to_stock.get(&key).copied()
95    }
96
97    /// Read the retained last-good row without changing freshness state.
98    /// Runtime write paths must use [`Self::lookup`] so dirty data cannot be
99    /// mistaken for fresh; this snapshot is for diagnostics and failure tests.
100    #[must_use]
101    pub fn snapshot_item(&self, stock_id: u64, key: i64) -> Option<PriceReminderCachedItem> {
102        self.inner
103            .lock()
104            .by_stock
105            .get(&stock_id)
106            .and_then(|items| items.get(&key))
107            .cloned()
108    }
109
110    pub fn mark_stock_dirty(&self, stock_id: u64) {
111        self.inner.lock().dirty_stocks.insert(stock_id);
112    }
113
114    pub fn record_server_seq(&self, server_seq: i32) {
115        self.inner.lock().change_state.record_server_seq(server_seq);
116    }
117
118    #[must_use]
119    pub fn handle_change_push(
120        &self,
121        stock_id: i64,
122        server_seq: i32,
123    ) -> PriceReminderChangePushAction {
124        let mut inner = self.inner.lock();
125        let action = inner.change_state.handle_change_push(stock_id, server_seq);
126        if let PriceReminderChangePushAction::MarkStockDirty { stock_id } = action {
127            inner.dirty_stocks.insert(stock_id);
128        }
129        action
130    }
131
132    pub fn apply_set_action(&self, action: PriceReminderSetCacheAction) {
133        self.apply_set_action_at(action, Instant::now());
134    }
135
136    pub fn apply_set_action_at(&self, action: PriceReminderSetCacheAction, now: Instant) {
137        let mut inner = self.inner.lock();
138        match action {
139            PriceReminderSetCacheAction::Upsert(item) => upsert_item(&mut inner, item, now),
140            PriceReminderSetCacheAction::RemoveKey { key } => remove_key(&mut inner, key),
141            PriceReminderSetCacheAction::ClearStock { stock_id } => {
142                remove_stock_items(&mut inner, stock_id);
143            }
144            PriceReminderSetCacheAction::MarkDirty { stock_id } => {
145                inner.dirty_stocks.insert(stock_id);
146            }
147        }
148    }
149
150    pub fn clear_idle_at(&self, now: Instant) -> usize {
151        let mut inner = self.inner.lock();
152        let expired = inner
153            .last_access
154            .iter()
155            .filter_map(|(stock_id, accessed_at)| {
156                now.checked_duration_since(*accessed_at)
157                    .filter(|idle| *idle > PRICE_REMINDER_IDLE_TTL)
158                    .map(|_| *stock_id)
159            })
160            .collect::<HashSet<_>>();
161
162        for stock_id in &expired {
163            inner.by_stock.remove(stock_id);
164            inner.dirty_stocks.remove(stock_id);
165            inner.last_access.remove(stock_id);
166        }
167        inner
168            .key_to_stock
169            .retain(|_, stock_id| !expired.contains(stock_id));
170        expired.len()
171    }
172}
173
174fn upsert_item(inner: &mut PriceReminderStateInner, item: PriceReminderCachedItem, now: Instant) {
175    if item.stock_id == 0 || item.key == 0 {
176        return;
177    }
178    let stock_id = item.stock_id;
179    let key = item.key;
180    remove_key_from_previous_stock(inner, key, stock_id);
181    inner
182        .by_stock
183        .entry(stock_id)
184        .or_default()
185        .insert(key, item);
186    inner.key_to_stock.insert(key, stock_id);
187    inner.last_access.insert(stock_id, now);
188}
189
190fn remove_key(inner: &mut PriceReminderStateInner, key: i64) {
191    if key == 0 {
192        return;
193    }
194    let Some(stock_id) = inner.key_to_stock.remove(&key) else {
195        return;
196    };
197    if let Some(items) = inner.by_stock.get_mut(&stock_id) {
198        items.remove(&key);
199    }
200}
201
202fn remove_stock_items(inner: &mut PriceReminderStateInner, stock_id: u64) {
203    inner.by_stock.remove(&stock_id);
204    inner
205        .key_to_stock
206        .retain(|_, mapped_stock_id| *mapped_stock_id != stock_id);
207}
208
209fn remove_key_from_previous_stock(
210    inner: &mut PriceReminderStateInner,
211    key: i64,
212    new_stock_id: u64,
213) {
214    let Some(previous_stock_id) = inner.key_to_stock.get(&key).copied() else {
215        return;
216    };
217    if previous_stock_id == new_stock_id {
218        return;
219    }
220    if let Some(previous_items) = inner.by_stock.get_mut(&previous_stock_id) {
221        previous_items.remove(&key);
222    }
223}
224
225#[cfg(test)]
226mod tests;
futu_cache/price_reminder_state.rs

futu_cache/
price_reminder_state.rs
