futu_gateway_core/bridge/
qot_right_push.rs

1//! QOT entitlement-change push handling.
2//!
3//! Mechanical split from `qot_right_refresh.rs`: active entitlement requests stay
4//! in the refresh module, while backend push updates (CMD6006 / CMD6651) live
5//! here.
6
7use std::sync::{Arc, Mutex, OnceLock};
8use std::time::{Duration, Instant};
9
10use futu_cache::qot_right::{QotRightBackendUpdate, QotRightCache};
11
12use super::push_delivery::try_send_qot_broadcast_with_metric;
13use super::{GatewayBridge, PushEvent};
14
15#[derive(Clone)]
16pub struct QotRightPushDedup {
17    inner: Arc<Mutex<QotRightPushDedupState>>,
18}
19
20#[derive(Default)]
21struct QotRightPushDedupState {
22    last_hash: Option<u64>,
23    last_at: Option<Instant>,
24}
25
26impl QotRightPushDedup {
27    pub fn new() -> Self {
28        Self {
29            inner: Arc::new(Mutex::new(QotRightPushDedupState::default())),
30        }
31    }
32
33    fn should_emit(&self, hash: u64, source: &'static str) -> bool {
34        let now = Instant::now();
35        match self.inner.lock() {
36            Ok(mut state) => state.should_emit(hash, now),
37            Err(poisoned) => {
38                tracing::warn!(
39                    source,
40                    "qot_right push dedup lock poisoned; recovering state"
41                );
42                let mut state = poisoned.into_inner();
43                state.should_emit(hash, now)
44            }
45        }
46    }
47}
48
49impl Default for QotRightPushDedup {
50    fn default() -> Self {
51        Self::new()
52    }
53}
54
55impl QotRightPushDedupState {
56    fn should_emit(&mut self, hash: u64, now: Instant) -> bool {
57        const DEDUP_WINDOW: Duration = Duration::from_secs(30);
58        match (self.last_hash, self.last_at) {
59            (Some(h), Some(t)) if h == hash && now.duration_since(t) < DEDUP_WINDOW => false,
60            _ => {
61                self.last_hash = Some(hash);
62                self.last_at = Some(now);
63                true
64            }
65        }
66    }
67}
68
69pub(super) fn qot_right_push_dedup_cache() -> QotRightPushDedup {
70    static CACHE: OnceLock<QotRightPushDedup> = OnceLock::new();
71    CACHE.get_or_init(QotRightPushDedup::new).clone()
72}
73
74fn should_emit_qot_right_push(dedup: &QotRightPushDedup, hash: u64, source: &'static str) -> bool {
75    dedup.should_emit(hash, source)
76}
77
78#[cfg(test)]
79#[path = "qot_right_push_tests.rs"]
80mod qot_right_push_tests;
81
82fn build_notify_qot_right_response(
83    data: &futu_cache::qot_right::QotRightData,
84) -> futu_proto::notify::Response {
85    use futu_proto::notify;
86    let qot_right = notify::QotRight {
87        hk_qot_right: data.hk_qot_right,
88        us_qot_right: data.api_us_qot_right,
89        cn_qot_right: data.sh_qot_right,
90        hk_option_qot_right: Some(data.hk_option_qot_right),
91        has_us_option_qot_right: Some(data.has_us_option_qot_right),
92        hk_future_qot_right: Some(data.hk_future_qot_right),
93        us_future_qot_right: Some(data.us_cme_future_qot_right),
94        us_option_qot_right: Some(data.us_option_qot_right),
95        us_index_qot_right: Some(data.us_index_qot_right),
96        us_otc_qot_right: Some(data.us_otc_qot_right),
97        sg_future_qot_right: Some(data.sg_future_qot_right),
98        jp_future_qot_right: Some(data.jp_future_qot_right),
99        us_cme_future_qot_right: Some(data.us_cme_future_qot_right),
100        us_cbot_future_qot_right: Some(data.us_cbot_future_qot_right),
101        us_nymex_future_qot_right: Some(data.us_nymex_future_qot_right),
102        us_comex_future_qot_right: Some(data.us_comex_future_qot_right),
103        us_cboe_future_qot_right: Some(data.us_cboe_future_qot_right),
104        sh_qot_right: Some(data.sh_qot_right),
105        sz_qot_right: Some(data.sz_qot_right),
106        cc_qot_right: Some(data.cc_qot_right),
107        sg_stock_qot_right: Some(data.sg_stock_qot_right),
108        my_stock_qot_right: Some(data.my_stock_qot_right),
109        jp_stock_qot_right: Some(data.jp_stock_qot_right),
110    };
111    let s2c = notify::S2c {
112        r#type: 4,
113        event: None,
114        program_status: None,
115        connect_status: None,
116        qot_right: Some(qot_right),
117        api_level: None,
118        api_quota: None,
119        used_quota: None,
120    };
121    notify::Response {
122        ret_type: 0,
123        ret_msg: None,
124        err_code: None,
125        s2c: Some(s2c),
126    }
127}
128
129impl GatewayBridge {
130    pub fn handle_qot_right_push_6651(
131        body: &[u8],
132        qot_right_cache: &QotRightCache,
133        dedup: &QotRightPushDedup,
134        push_tx: &tokio::sync::mpsc::Sender<PushEvent>,
135        metrics: &futu_server::metrics::GatewayMetrics,
136    ) -> bool {
137        use futu_backend::proto_internal::ftcmd6651_qta_auth_chg as chg;
138        use std::hash::{Hash, Hasher};
139
140        let pb: chg::QuoteChangeNotify = match prost::Message::decode(body) {
141            Ok(m) => m,
142            Err(e) => {
143                tracing::warn!(
144                    body_len = body.len(),
145                    error = %e,
146                    "CMD6651 decode failed"
147                );
148                return false;
149            }
150        };
151
152        let items: Vec<(i32, i32, i32)> = pb
153            .quote_change_list
154            .iter()
155            .map(|it| {
156                (
157                    it.quote_type.unwrap_or(0),
158                    it.quote_before_change.unwrap_or(0),
159                    it.quote_after_change.unwrap_or(0),
160                )
161            })
162            .collect();
163        qot_right_cache.set_pushed_quote_change_notify(
164            futu_cache::qot_right::PushedQuoteChangeNotify {
165                change_items: items.clone(),
166            },
167        );
168
169        let changed_types = qot_right_cache.apply_6651_changes(&items);
170        tracing::info!(
171            change_count = items.len(),
172            applied_types = ?changed_types,
173            "CMD6651 QuoteChangeNotify applied"
174        );
175
176        if !changed_types.is_empty() {
177            tracing::warn!(
178                applied_types = ?changed_types,
179                "qot_right changed via CMD6651; pushed notify cached, in-memory rights applied, \
180                 NotifyType_QotRight broadcast will be attempted"
181            );
182        }
183
184        let data = qot_right_cache.get();
185        let mut hasher = std::collections::hash_map::DefaultHasher::new();
186        data.hk_qot_right.hash(&mut hasher);
187        data.api_us_qot_right.hash(&mut hasher);
188        data.sh_qot_right.hash(&mut hasher);
189        data.sz_qot_right.hash(&mut hasher);
190        data.hk_option_qot_right.hash(&mut hasher);
191        data.hk_future_qot_right.hash(&mut hasher);
192        data.has_us_option_qot_right.hash(&mut hasher);
193        data.us_option_qot_right.hash(&mut hasher);
194        data.us_index_qot_right.hash(&mut hasher);
195        data.us_otc_qot_right.hash(&mut hasher);
196        data.us_cme_future_qot_right.hash(&mut hasher);
197        data.us_cbot_future_qot_right.hash(&mut hasher);
198        data.us_nymex_future_qot_right.hash(&mut hasher);
199        data.us_comex_future_qot_right.hash(&mut hasher);
200        data.us_cboe_future_qot_right.hash(&mut hasher);
201        data.sg_future_qot_right.hash(&mut hasher);
202        data.sg_stock_qot_right.hash(&mut hasher);
203        data.my_stock_qot_right.hash(&mut hasher);
204        data.jp_future_qot_right.hash(&mut hasher);
205        data.jp_stock_qot_right.hash(&mut hasher);
206        data.cc_qot_right.hash(&mut hasher);
207        let hash = hasher.finish();
208
209        let should_broadcast = should_emit_qot_right_push(dedup, hash, "CMD6651");
210
211        if should_broadcast {
212            let notify_resp = build_notify_qot_right_response(&data);
213            let body_bytes = prost::Message::encode_to_vec(&notify_resp);
214            let proto_id = futu_core::proto_id::NOTIFY;
215            try_send_qot_broadcast_with_metric(
216                push_tx,
217                proto_id,
218                body_bytes,
219                metrics,
220                "CMD6651 qot_right broadcast",
221            );
222            tracing::info!(hash, "CMD6651: NotifyType_QotRight broadcast attempted");
223        }
224
225        true
226    }
227
228    pub fn handle_qot_right_push_6006(
229        body: &[u8],
230        qot_right_cache: &QotRightCache,
231        dedup: &QotRightPushDedup,
232        push_tx: &tokio::sync::mpsc::Sender<PushEvent>,
233        metrics: &futu_server::metrics::GatewayMetrics,
234    ) -> bool {
235        use futu_backend::proto_internal::ftcmd6006_qta_auth_chg as chg;
236        use std::hash::{Hash, Hasher};
237
238        let pb: chg::QtaAuthChgNotify = match prost::Message::decode(body) {
239            Ok(m) => m,
240            Err(e) => {
241                tracing::warn!(body_len = body.len(), error = %e, "CMD6006 decode failed");
242                return false;
243            }
244        };
245
246        // Ref: proto-internal/FTCMD6006_QtaAuthChg.proto:7-9.
247        // Missing field or server value 0 means "discard and keep previous right".
248        // quote_flag_* / quote_diff_rsn_* are not projected by current QotRightData.
249        let items: Vec<(i32, u32)> = [
250            qot_right_6006_auth_item(1, pb.quote_auth_hk),
251            qot_right_6006_auth_item(3, pb.quote_auth_us),
252        ]
253        .into_iter()
254        .flatten()
255        .collect();
256
257        let mut applied: Vec<i32> = Vec::new();
258        let hk_us_changed = qot_right_cache.apply_direct_auth_changes(&items);
259        applied.extend(hk_us_changed);
260
261        if let Some(cn_val) = pb.quote_auth_cn.filter(|v| *v != 0) {
262            qot_right_cache.update_from_backend(QotRightBackendUpdate {
263                cn_got: Some(cn_val),
264                ..Default::default()
265            });
266            applied.push(2);
267        }
268
269        tracing::info!(applied_types = ?applied, "CMD6006 applied");
270
271        if !applied.is_empty() {
272            tracing::warn!(
273                applied_types = ?applied,
274                "qot_right changed via CMD6006; in-memory rights applied, \
275                 NotifyType_QotRight broadcast will be attempted"
276            );
277        }
278
279        let data = qot_right_cache.get();
280        let mut hasher = std::collections::hash_map::DefaultHasher::new();
281        data.hk_qot_right.hash(&mut hasher);
282        data.api_us_qot_right.hash(&mut hasher);
283        data.sh_qot_right.hash(&mut hasher);
284        data.sz_qot_right.hash(&mut hasher);
285        data.sg_stock_qot_right.hash(&mut hasher);
286        data.my_stock_qot_right.hash(&mut hasher);
287        data.jp_stock_qot_right.hash(&mut hasher);
288        data.cc_qot_right.hash(&mut hasher);
289        let hash = hasher.finish();
290
291        let should_broadcast = should_emit_qot_right_push(dedup, hash, "CMD6006");
292
293        if should_broadcast {
294            let notify_resp = build_notify_qot_right_response(&data);
295            let body_bytes = prost::Message::encode_to_vec(&notify_resp);
296            let proto_id = futu_core::proto_id::NOTIFY;
297            try_send_qot_broadcast_with_metric(
298                push_tx,
299                proto_id,
300                body_bytes,
301                metrics,
302                "CMD6006 qot_right broadcast",
303            );
304        }
305
306        true
307    }
308}
309
310fn qot_right_6006_auth_item(quote_type: i32, value: Option<u32>) -> Option<(i32, u32)> {
311    value.filter(|v| *v != 0).map(|v| (quote_type, v))
312}
futu_gateway_core/bridge/qot_right_push.rs

futu_gateway_core/bridge/
qot_right_push.rs
