1use super::{
2 CachedSecurityInfo, CryptoPairInfo, CryptoTradeConfig, OnDemandSecurityPublishOutcome,
3 OptionContractInfo, StaticDataCache,
4};
5use std::collections::HashSet;
6use std::sync::Arc;
7
8impl StaticDataCache {
9 pub fn upsert_full_security_info(&self, key: &str, info: CachedSecurityInfo) {
20 debug_assert!(
21 info.source.is_complete(),
22 "upsert_full_security_info called with non-complete source ({:?})",
23 info.source
24 );
25 let _identity_write = match self.zero_id_repair.lock() {
26 Ok(guard) => guard,
27 Err(_) => return,
28 };
29 self.upsert_with_owner_index_maintenance(key, info);
30 }
31
32 pub fn upsert_full_security_info_with_crypto_pair(
35 &self,
36 key: &str,
37 info: CachedSecurityInfo,
38 pair: CryptoPairInfo,
39 ) {
40 debug_assert!(
41 info.source.is_complete(),
42 "upsert_full_security_info_with_crypto_pair called with non-complete source ({:?})",
43 info.source
44 );
45 let _identity_write = match self.zero_id_repair.lock() {
46 Ok(guard) => guard,
47 Err(_) => return,
48 };
49 let stock_id = info.stock_id;
50 if stock_id > 0 {
51 self.upsert_crypto_pair_by_stock_id_locked(stock_id, pair);
52 self.upsert_with_owner_index_maintenance(key, info);
53 } else {
54 self.upsert_with_owner_index_maintenance(key, info);
55 self.upsert_crypto_pair_info_locked(key, pair);
56 }
57 }
58
59 pub fn upsert_crypto_pair_info(&self, key: &str, pair: CryptoPairInfo) {
61 let _identity_write = match self.zero_id_repair.lock() {
62 Ok(guard) => guard,
63 Err(_) => return,
64 };
65 self.upsert_crypto_pair_info_locked(key, pair);
66 }
67
68 fn upsert_crypto_pair_info_locked(&self, key: &str, pair: CryptoPairInfo) {
69 if let Some(stock_id) = self
70 .securities
71 .get(key)
72 .map(|entry| entry.stock_id)
73 .filter(|stock_id| *stock_id > 0)
74 {
75 self.upsert_crypto_pair_by_stock_id_locked(stock_id, pair.clone());
76 }
77 if pair.cc_origin.is_empty() && pair.cc_destination.is_empty() {
78 self.crypto_pairs.remove(key);
79 } else {
80 self.crypto_pairs.insert(key.to_string(), pair);
81 }
82 }
83
84 fn upsert_crypto_pair_by_stock_id_locked(&self, stock_id: u64, pair: CryptoPairInfo) {
85 if stock_id == 0 {
86 return;
87 }
88 if pair.cc_origin.is_empty() && pair.cc_destination.is_empty() {
89 self.crypto_pairs_by_stock_id.remove(&stock_id);
90 } else {
91 self.crypto_pairs_by_stock_id.insert(stock_id, pair);
92 }
93 }
94
95 fn refresh_public_crypto_pair_locked(&self, key: &str) {
96 let pair = self.securities.get(key).and_then(|security| {
97 (security.stock_id > 0)
98 .then_some(security.stock_id)
99 .and_then(|stock_id| {
100 self.crypto_pairs_by_stock_id
101 .get(&stock_id)
102 .map(|pair| pair.clone())
103 })
104 });
105 if let Some(pair) = pair {
106 self.crypto_pairs.insert(key.to_string(), pair);
107 } else {
108 self.crypto_pairs.remove(key);
109 }
110 }
111
112 pub fn set_option_contract_info(&self, stock_id: u64, info: OptionContractInfo) {
114 if stock_id == 0 {
115 return;
116 }
117 self.option_contracts.insert(stock_id, info);
118 }
119
120 pub fn get_option_contract_info_by_stock_id(
122 &self,
123 stock_id: u64,
124 ) -> Option<OptionContractInfo> {
125 self.option_contracts
126 .get(&stock_id)
127 .map(|entry| *entry.value())
128 }
129
130 pub fn get_crypto_pair_info(&self, key: &str) -> Option<CryptoPairInfo> {
132 self.crypto_pairs.get(key).map(|v| v.clone())
133 }
134
135 fn crypto_trade_config_key(broker_id: u32, symbol: &str, exchange: &str) -> String {
136 format!(
137 "{broker_id}:{}:{}",
138 symbol.trim().to_ascii_uppercase(),
139 exchange.trim().to_ascii_uppercase()
140 )
141 }
142
143 pub fn set_crypto_trade_configs_for_broker(
145 &self,
146 broker_id: u32,
147 configs: Vec<CryptoTradeConfig>,
148 ) {
149 let prefix = format!("{broker_id}:");
150 self.crypto_trade_configs
151 .retain(|key, _| !key.starts_with(&prefix));
152 for config in configs {
153 if config.symbol.trim().is_empty() || config.exchange.trim().is_empty() {
154 continue;
155 }
156 let key = Self::crypto_trade_config_key(broker_id, &config.symbol, &config.exchange);
157 self.crypto_trade_configs.insert(key, config);
158 }
159 }
160
161 pub fn get_crypto_trade_config(
163 &self,
164 broker_id: u32,
165 symbol: &str,
166 exchange: &str,
167 ) -> Option<CryptoTradeConfig> {
168 let key = Self::crypto_trade_config_key(broker_id, symbol, exchange);
169 self.crypto_trade_configs.get(&key).map(|v| v.clone())
170 }
171
172 pub fn upsert_basic_security_info(&self, key: &str, info: CachedSecurityInfo) {
178 debug_assert!(
179 !info.source.is_complete(),
180 "upsert_basic_security_info called with complete source ({:?}); use upsert_full",
181 info.source
182 );
183 debug_assert_eq!(
184 info.warrnt_stock_owner, 0,
185 "OnDemandBasic must have warrnt_stock_owner=0 (caller didn't query the field)"
186 );
187 let _identity_write = match self.zero_id_repair.lock() {
188 Ok(guard) => guard,
189 Err(_) => return,
190 };
191 self.upsert_basic_security_info_locked(key, info);
192 }
193
194 pub fn upsert_basic_security_info_by_stock_id_only(&self, info: CachedSecurityInfo) {
201 debug_assert!(
202 !info.source.is_complete(),
203 "ID-only on-demand rows must not replace complete stock-list rows"
204 );
205 if info.stock_id == 0 {
206 return;
207 }
208 let _identity_write = match self.zero_id_repair.lock() {
209 Ok(guard) => guard,
210 Err(_) => return,
211 };
212 if self
213 .securities_by_stock_id
214 .get(&info.stock_id)
215 .is_some_and(|existing| existing.is_complete() || !existing.code.is_empty())
216 {
217 return;
218 }
219 self.securities_by_stock_id
220 .insert(info.stock_id, Arc::new(info));
221 }
222
223 fn upsert_basic_security_info_locked(&self, key: &str, info: CachedSecurityInfo) {
224 if let Some(existing) = self.securities.get(key)
230 && existing.is_complete()
231 {
232 tracing::debug!(
233 key,
234 "upsert_basic_security_info skipped: existing complete row prevails"
235 );
236 return;
237 }
238 if info.stock_id > 0
239 && self
240 .securities_by_stock_id
241 .get(&info.stock_id)
242 .is_some_and(|existing| existing.is_complete())
243 {
244 tracing::debug!(
245 key,
246 stock_id = info.stock_id,
247 "upsert_basic_security_info skipped: exact complete row prevails"
248 );
249 return;
250 }
251 self.upsert_with_owner_index_maintenance(key, info);
252 }
253
254 pub fn publish_on_demand_security_info(
260 &self,
261 key: &str,
262 refreshed: CachedSecurityInfo,
263 crypto_pair: Option<CryptoPairInfo>,
264 ) -> OnDemandSecurityPublishOutcome {
265 self.publish_on_demand_security_info_with_pair_hook(key, refreshed, crypto_pair, || {})
266 }
267
268 pub(super) fn publish_on_demand_security_info_with_pair_hook(
269 &self,
270 key: &str,
271 refreshed: CachedSecurityInfo,
272 crypto_pair: Option<CryptoPairInfo>,
273 before_pair_publish: impl FnOnce(),
274 ) -> OnDemandSecurityPublishOutcome {
275 let _identity_write = match self.zero_id_repair.lock() {
276 Ok(guard) => guard,
277 Err(_) => return OnDemandSecurityPublishOutcome::Rejected,
278 };
279 if refreshed.stock_id == 0
280 || refreshed.source.is_complete()
281 || refreshed.warrnt_stock_owner != 0
282 {
283 return OnDemandSecurityPublishOutcome::Rejected;
284 }
285
286 let stock_id = refreshed.stock_id;
287 if self
288 .id_to_key
289 .get(&stock_id)
290 .is_some_and(|mapped| mapped.as_str() != key)
291 {
292 return OnDemandSecurityPublishOutcome::Rejected;
293 }
294
295 let existing = self
296 .securities
297 .get(key)
298 .map(|entry| Arc::clone(entry.value()));
299 match existing {
300 Some(existing) if existing.stock_id == 0 => {
301 let stock_id = refreshed.stock_id;
302 if self.repair_zero_stock_id_from_basic_locked(key, refreshed, || {}) {
303 if let Some(pair) = crypto_pair {
304 self.upsert_crypto_pair_by_stock_id_locked(stock_id, pair);
305 self.refresh_public_crypto_pair_locked(key);
306 }
307 OnDemandSecurityPublishOutcome::ZeroIdRepaired
308 } else {
309 OnDemandSecurityPublishOutcome::Rejected
310 }
311 }
312 Some(existing) if existing.stock_id != stock_id => {
313 OnDemandSecurityPublishOutcome::Rejected
314 }
315 Some(existing) if existing.is_complete() => {
316 if self.update_mkt_id_locked(key, refreshed.mkt_id) {
317 OnDemandSecurityPublishOutcome::CompleteMktIdUpdated
318 } else {
319 OnDemandSecurityPublishOutcome::Rejected
320 }
321 }
322 Some(_) | None => {
323 let stock_id = refreshed.stock_id;
324 self.upsert_basic_security_info_locked(key, refreshed);
325 before_pair_publish();
326 if let Some(pair) = crypto_pair {
327 self.upsert_crypto_pair_by_stock_id_locked(stock_id, pair);
328 self.refresh_public_crypto_pair_locked(key);
329 }
330 OnDemandSecurityPublishOutcome::BasicPublished
331 }
332 }
333 }
334
335 pub fn repair_zero_stock_id_from_basic(
343 &self,
344 key: &str,
345 refreshed: CachedSecurityInfo,
346 ) -> bool {
347 self.repair_zero_stock_id_from_basic_with_security_hook(key, refreshed, || {})
348 }
349
350 pub(super) fn repair_zero_stock_id_from_basic_with_security_hook(
351 &self,
352 key: &str,
353 refreshed: CachedSecurityInfo,
354 after_security_locked: impl FnOnce(),
355 ) -> bool {
356 let _repair = match self.zero_id_repair.lock() {
361 Ok(guard) => guard,
362 Err(_) => return false,
363 };
364
365 self.repair_zero_stock_id_from_basic_locked(key, refreshed, after_security_locked)
366 }
367
368 fn repair_zero_stock_id_from_basic_locked(
369 &self,
370 key: &str,
371 refreshed: CachedSecurityInfo,
372 after_security_locked: impl FnOnce(),
373 ) -> bool {
374 if refreshed.stock_id == 0
375 || refreshed.source.is_complete()
376 || refreshed.warrnt_stock_owner != 0
377 {
378 return false;
379 }
380
381 let stock_id = refreshed.stock_id;
382 let old_public_keys = self.public_keys_for_stock_id_locked(stock_id);
383 if self
384 .id_to_key
385 .get(&stock_id)
386 .is_some_and(|mapped| mapped.as_str() != key)
387 {
388 return false;
389 }
390
391 let Some(preliminary_ref) = self.securities.get(key) else {
392 return false;
393 };
394 let preliminary = Arc::clone(preliminary_ref.value());
395 drop(preliminary_ref);
396 if preliminary.stock_id != 0
397 || preliminary.market != refreshed.market
398 || preliminary.code != refreshed.code
399 {
400 return false;
401 }
402
403 let preliminary_owner = preliminary.warrnt_stock_owner;
408 let mut owner_index = if preliminary_owner != 0 {
409 match self.owner_to_warrants.write() {
410 Ok(index) => Some(index),
411 Err(_) => return false,
412 }
413 } else {
414 None
415 };
416
417 let Some(mut entry) = self.securities.get_mut(key) else {
418 return false;
419 };
420 after_security_locked();
421 let old = Arc::clone(entry.value());
422 if old.stock_id != 0
423 || old.market != refreshed.market
424 || old.code != refreshed.code
425 || old.warrnt_stock_owner != preliminary_owner
426 {
427 return false;
428 }
429
430 let mut updated = if old.is_complete() {
431 old.as_ref().clone()
432 } else {
433 refreshed.clone()
434 };
435 updated.stock_id = stock_id;
436 updated.mkt_id = refreshed.mkt_id;
437
438 let old_owner = old.warrnt_stock_owner;
439 let new_owner = updated.warrnt_stock_owner;
440 debug_assert_eq!(old_owner, preliminary_owner);
441 debug_assert!(new_owner == 0 || owner_index.is_some());
442
443 let stock_id_reservation = match self.id_to_key.entry(stock_id) {
448 dashmap::mapref::entry::Entry::Occupied(occupied) => {
449 if occupied.get() != key {
450 return false;
451 }
452 occupied.into_ref()
453 }
454 dashmap::mapref::entry::Entry::Vacant(vacant) => vacant.insert(key.to_string()),
455 };
456
457 self.remove_future_main_link_aliases(key, &old);
458 let updated = Arc::new(updated);
459 *entry = Arc::clone(&updated);
460 self.securities_by_stock_id
461 .insert(stock_id, Arc::clone(&updated));
462 if let Some(pair) = self.crypto_pairs.get(key).map(|pair| pair.clone()) {
463 self.crypto_pairs_by_stock_id.insert(stock_id, pair);
464 }
465 self.add_future_main_link_aliases(key, &updated);
466 drop(entry);
467
468 if let Some(index) = owner_index.as_mut() {
469 if old_owner != 0
470 && !self.securities.iter().any(|candidate| {
471 candidate.stock_id == 0 && candidate.warrnt_stock_owner == old_owner
472 })
473 && let Some(warrants) = index.get_mut(&old_owner)
474 {
475 warrants.remove(&0);
476 if warrants.is_empty() {
477 index.remove(&old_owner);
478 }
479 }
480 if new_owner != 0 {
481 index.entry(new_owner).or_default().insert(stock_id);
482 }
483 }
484
485 drop(stock_id_reservation);
486 self.id_to_key.remove_if(&0, |_, mapped| mapped == key);
487 let new_public_keys = self.public_keys_for_stock_id_locked(stock_id);
488 self.reconcile_public_stock_id_keys_locked(stock_id, &old_public_keys, &new_public_keys);
489 true
490 }
491
492 pub fn delete_security_info(&self, stock_id: u64) -> bool {
498 let _identity_write = match self.zero_id_repair.lock() {
499 Ok(guard) => guard,
500 Err(_) => return false,
501 };
502 let old_public_keys = self.public_keys_for_stock_id_locked(stock_id);
503 let Some((_, key)) = self.id_to_key.remove(&stock_id) else {
504 return false;
505 };
506 self.option_contracts.remove(&stock_id);
507 if stock_id == 0 {
508 if let Some((_, old)) = self
509 .securities
510 .remove_if(&key, |_, info| info.stock_id == 0)
511 {
512 self.remove_future_main_link_aliases(&key, &old);
513 self.remove_owner_relation(&old);
514 }
515 self.crypto_pairs.remove(&key);
516 self.stale_mkt_ids.remove(&key);
517 return true;
518 }
519
520 self.security_request_aliases.remove(&stock_id);
521
522 if let Some((_, old)) = self.securities_by_stock_id.remove(&stock_id) {
523 self.remove_owner_relation(&old);
524 self.remove_future_main_link_aliases_if_unreferenced(&key, &old);
525 }
526 self.crypto_pairs_by_stock_id.remove(&stock_id);
527 self.reconcile_public_stock_id_keys_locked(stock_id, &old_public_keys, &HashSet::new());
528 for public_key in old_public_keys {
529 self.refresh_public_security_locked(&public_key);
530 }
531
532 if let Ok(mut map) = self.owner_to_warrants.write() {
534 map.remove(&stock_id);
535 }
536 true
537 }
538
539 fn upsert_with_owner_index_maintenance(&self, key: &str, info: CachedSecurityInfo) {
541 let stock_id = info.stock_id;
542 if stock_id == 0 {
543 self.upsert_zero_id_security_locked(key, info);
544 return;
545 }
546
547 let old_public_keys = self.public_keys_for_stock_id_locked(stock_id);
548 let old_key = self.id_to_key.get(&stock_id).map(|mapped| mapped.clone());
549 let old_exact = self
550 .securities_by_stock_id
551 .remove(&stock_id)
552 .map(|(_, old)| old);
553 self.id_to_key.remove(&stock_id);
554
555 if let (Some(old_key), Some(old)) = (old_key.as_deref(), old_exact.as_ref())
556 && !old.is_complete()
557 && old_key != key
558 {
559 self.security_request_aliases
560 .entry(stock_id)
561 .or_default()
562 .insert(old_key.to_string());
563 }
564 self.remove_security_request_alias_locked(stock_id, key);
565
566 if let (Some(old_key), Some(old)) = (old_key.as_deref(), old_exact.as_ref()) {
567 self.remove_owner_relation(old);
568 self.remove_future_main_link_aliases_if_unreferenced(old_key, old);
569 }
570
571 self.remove_zero_id_public_row_locked(key);
572
573 let info = Arc::new(info);
574 self.securities_by_stock_id
575 .insert(stock_id, Arc::clone(&info));
576 self.id_to_key.insert(stock_id, key.to_string());
577 self.add_owner_relation(&info);
578 self.add_future_main_link_aliases(key, &info);
579
580 let new_public_keys = self.public_keys_for_stock_id_locked(stock_id);
581 self.reconcile_public_stock_id_keys_locked(stock_id, &old_public_keys, &new_public_keys);
582 let mut public_keys_to_refresh = old_public_keys;
583 public_keys_to_refresh.extend(new_public_keys);
584 for public_key in public_keys_to_refresh {
585 self.refresh_public_security_locked(&public_key);
586 }
587 }
588
589 fn upsert_zero_id_security_locked(&self, key: &str, info: CachedSecurityInfo) {
590 if self.public_security_candidate_locked(key).is_some() {
591 return;
592 }
593 self.remove_zero_id_public_row_locked(key);
594 let info = Arc::new(info);
595 self.securities.insert(key.to_string(), Arc::clone(&info));
596 self.id_to_key.insert(0, key.to_string());
597 self.add_owner_relation(&info);
598 self.add_future_main_link_aliases(key, &info);
599 }
600
601 fn remove_zero_id_public_row_locked(&self, key: &str) {
602 let Some((_, old)) = self.securities.remove_if(key, |_, info| info.stock_id == 0) else {
603 return;
604 };
605 self.id_to_key.remove_if(&0, |_, mapped| mapped == key);
606 self.remove_future_main_link_aliases(key, &old);
607 self.remove_owner_relation(&old);
608 }
609
610 fn public_security_candidate_locked(&self, key: &str) -> Option<Arc<CachedSecurityInfo>> {
615 let stock_ids = self.public_stock_ids_by_key.get(key)?;
616 stock_ids.iter().find_map(|stock_id| {
617 #[cfg(test)]
618 self.public_security_candidate_inspections
619 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
620 self.securities_by_stock_id
621 .get(stock_id)
622 .map(|info| Arc::clone(info.value()))
623 .filter(|info| !info.no_search)
624 })
625 }
626
627 fn public_keys_for_stock_id_locked(&self, stock_id: u64) -> HashSet<String> {
628 if stock_id == 0 {
629 return HashSet::new();
630 }
631 let mut keys = self.security_request_alias_keys_locked(stock_id);
632 if let Some(canonical) = self.id_to_key.get(&stock_id) {
633 keys.insert(canonical.clone());
634 }
635 keys
636 }
637
638 fn reconcile_public_stock_id_keys_locked(
639 &self,
640 stock_id: u64,
641 old_keys: &HashSet<String>,
642 new_keys: &HashSet<String>,
643 ) {
644 debug_assert!(stock_id > 0);
645 for key in old_keys.difference(new_keys) {
646 let remove_entry =
647 if let Some(mut stock_ids) = self.public_stock_ids_by_key.get_mut(key) {
648 stock_ids.remove(&stock_id);
649 stock_ids.is_empty()
650 } else {
651 false
652 };
653 if remove_entry {
654 self.public_stock_ids_by_key.remove(key);
655 }
656 }
657 for key in new_keys.difference(old_keys) {
658 self.public_stock_ids_by_key
659 .entry(key.clone())
660 .or_default()
661 .insert(stock_id);
662 }
663 }
664
665 #[cfg(test)]
666 pub(super) fn public_identity_index_matches_sources(&self) -> bool {
667 let mut expected =
668 std::collections::HashMap::<String, std::collections::BTreeSet<u64>>::new();
669 for mapped in &self.id_to_key {
670 if *mapped.key() > 0 {
671 expected
672 .entry(mapped.value().clone())
673 .or_default()
674 .insert(*mapped.key());
675 }
676 }
677 for aliases in &self.security_request_aliases {
678 for alias in aliases.value() {
679 expected
680 .entry(alias.clone())
681 .or_default()
682 .insert(*aliases.key());
683 }
684 }
685 let actual = self
686 .public_stock_ids_by_key
687 .iter()
688 .map(|entry| (entry.key().clone(), entry.value().clone()))
689 .collect::<std::collections::HashMap<_, _>>();
690 actual == expected
691 }
692
693 #[cfg(test)]
694 pub(super) fn reset_public_security_candidate_inspections(&self) {
695 self.public_security_candidate_inspections
696 .store(0, std::sync::atomic::Ordering::Relaxed);
697 }
698
699 #[cfg(test)]
700 pub(super) fn public_security_candidate_inspections(&self) -> u64 {
701 self.public_security_candidate_inspections
702 .load(std::sync::atomic::Ordering::Relaxed)
703 }
704
705 fn security_request_alias_keys_locked(&self, stock_id: u64) -> HashSet<String> {
706 self.security_request_aliases
707 .get(&stock_id)
708 .map(|aliases| aliases.clone())
709 .unwrap_or_default()
710 }
711
712 fn remove_security_request_alias_locked(&self, stock_id: u64, key: &str) {
713 let remove_entry =
714 if let Some(mut aliases) = self.security_request_aliases.get_mut(&stock_id) {
715 aliases.remove(key);
716 aliases.is_empty()
717 } else {
718 false
719 };
720 if remove_entry {
721 self.security_request_aliases.remove(&stock_id);
722 }
723 }
724
725 pub(super) fn refresh_public_keys_for_stock_id_locked(&self, stock_id: u64) {
726 for key in self.public_keys_for_stock_id_locked(stock_id) {
727 self.refresh_public_security_locked(&key);
728 }
729 }
730
731 fn refresh_public_security_locked(&self, key: &str) {
732 if let Some(candidate) = self.public_security_candidate_locked(key) {
733 if !candidate.needs_mkt_id_refresh() {
734 self.stale_mkt_ids.remove(key);
735 }
736 self.securities.insert(key.to_string(), candidate);
737 } else if self
738 .securities
739 .get(key)
740 .is_some_and(|current| current.stock_id > 0)
741 {
742 self.securities.remove(key);
743 self.stale_mkt_ids.remove(key);
744 }
745 self.refresh_public_crypto_pair_locked(key);
746 }
747}