futu_cache/static_data/

security_identity.rs

use super::{
    CachedSecurityInfo, CryptoPairInfo, CryptoTradeConfig, OnDemandSecurityPublishOutcome,
    OptionContractInfo, StaticDataCache,
};
use std::collections::HashSet;
use std::sync::Arc;

impl StaticDataCache {
    #[must_use]
    pub fn stock_list_publication_generation(&self) -> u64 {
        self.stock_list_publication_generation
            .load(std::sync::atomic::Ordering::Acquire)
    }

    /// v1.4.106 codex 1148 F9 (P3): 统一写入口 — 完整静态行 (`StockListFull` /
    /// `Bootstrap` source)。同步维护 `securities` + `id_to_key` + 若有 owner
    /// 还更新 `owner_to_warrants`。**自动 dedup**: 已有同 key 但 `warrnt_stock_owner`
    /// 变化时, 旧 owner 下移除该 warrant id, 新 owner 下添加。
    ///
    /// 替代生产代码里手动调 `securities.insert()` + `id_to_key.insert()` +
    /// `add_warrant_owner()` 三步骤的 pattern。
    ///
    /// **不允许** caller 把不完整的 source 标 `StockListFull`(若 `info.source ==
    /// OnDemandBasic`, 用 `upsert_basic_security_info` 而非本 fn)。
    pub fn upsert_full_security_info(&self, key: &str, info: CachedSecurityInfo) {
        debug_assert!(
            info.source.is_complete(),
            "upsert_full_security_info called with non-complete source ({:?})",
            info.source
        );
        let _identity_write = match self.zero_id_repair.lock() {
            Ok(guard) => guard,
            Err(_) => return,
        };
        self.upsert_with_owner_index_maintenance(key, info);
    }

    /// Atomically publish a complete stock-list row and its canonical crypto
    /// pair metadata under the same cache-owned identity transaction.
    pub fn upsert_full_security_info_with_crypto_pair(
        &self,
        key: &str,
        info: CachedSecurityInfo,
        pair: CryptoPairInfo,
    ) {
        debug_assert!(
            info.source.is_complete(),
            "upsert_full_security_info_with_crypto_pair called with non-complete source ({:?})",
            info.source
        );
        let _identity_write = match self.zero_id_repair.lock() {
            Ok(guard) => guard,
            Err(_) => return,
        };
        let stock_id = info.stock_id;
        if stock_id > 0 {
            self.upsert_crypto_pair_by_stock_id_locked(stock_id, pair);
            self.upsert_with_owner_index_maintenance(key, info);
        } else {
            self.upsert_with_owner_index_maintenance(key, info);
            self.upsert_crypto_pair_info_locked(key, pair);
        }
        self.stock_list_publication_generation
            .fetch_add(1, std::sync::atomic::Ordering::Release);
    }

    /// Publish a SQLite Bootstrap row only if no authoritative stock-list
    /// update/delete/clear linearized after the caller's DB-read snapshot.
    ///
    /// Existing complete exact rows always prevail. A canonical public key
    /// already owned by a different positive stock id is also rejected.
    pub fn publish_bootstrap_security_if_stock_list_generation(
        &self,
        expected_generation: u64,
        key: &str,
        info: CachedSecurityInfo,
    ) -> bool {
        if info.stock_id == 0
            || info.market <= 0
            || info.code.is_empty()
            || !info.source.is_complete()
        {
            return false;
        }
        let _identity_write = match self.zero_id_repair.lock() {
            Ok(guard) => guard,
            Err(_) => return false,
        };
        if self
            .stock_list_publication_generation
            .load(std::sync::atomic::Ordering::Acquire)
            != expected_generation
        {
            return false;
        }
        if self
            .securities_by_stock_id
            .get(&info.stock_id)
            .is_some_and(|existing| existing.is_complete())
        {
            return false;
        }
        if self
            .securities
            .get(key)
            .is_some_and(|existing| existing.stock_id > 0 && existing.stock_id != info.stock_id)
        {
            return false;
        }
        self.upsert_with_owner_index_maintenance(key, info);
        true
    }

    /// 写入 stock-list 下发的 crypto 货币对元数据。
    pub fn upsert_crypto_pair_info(&self, key: &str, pair: CryptoPairInfo) {
        let _identity_write = match self.zero_id_repair.lock() {
            Ok(guard) => guard,
            Err(_) => return,
        };
        self.upsert_crypto_pair_info_locked(key, pair);
    }

    fn upsert_crypto_pair_info_locked(&self, key: &str, pair: CryptoPairInfo) {
        if let Some(stock_id) = self
            .securities
            .get(key)
            .map(|entry| entry.stock_id)
            .filter(|stock_id| *stock_id > 0)
        {
            self.upsert_crypto_pair_by_stock_id_locked(stock_id, pair.clone());
        }
        if pair.cc_origin.is_empty() && pair.cc_destination.is_empty() {
            self.crypto_pairs.remove(key);
        } else {
            self.crypto_pairs.insert(key.to_string(), pair);
        }
    }

    fn upsert_crypto_pair_by_stock_id_locked(&self, stock_id: u64, pair: CryptoPairInfo) {
        if stock_id == 0 {
            return;
        }
        if pair.cc_origin.is_empty() && pair.cc_destination.is_empty() {
            self.crypto_pairs_by_stock_id.remove(&stock_id);
        } else {
            self.crypto_pairs_by_stock_id.insert(stock_id, pair);
        }
    }

    fn refresh_public_crypto_pair_locked(&self, key: &str) {
        let pair = self.securities.get(key).and_then(|security| {
            (security.stock_id > 0)
                .then_some(security.stock_id)
                .and_then(|stock_id| {
                    self.crypto_pairs_by_stock_id
                        .get(&stock_id)
                        .map(|pair| pair.clone())
                })
        });
        if let Some(pair) = pair {
            self.crypto_pairs.insert(key.to_string(), pair);
        } else {
            self.crypto_pairs.remove(key);
        }
    }

    /// Cache option contract metadata from CMD20106 `OptionResultInfo`.
    pub fn set_option_contract_info(&self, stock_id: u64, info: OptionContractInfo) {
        if stock_id == 0 {
            return;
        }
        self.option_contracts.insert(stock_id, info);
    }

    /// Read option contract metadata by option stock_id.
    pub fn get_option_contract_info_by_stock_id(
        &self,
        stock_id: u64,
    ) -> Option<OptionContractInfo> {
        self.option_contracts
            .get(&stock_id)
            .map(|entry| *entry.value())
    }

    /// 读取 crypto 货币对元数据。
    pub fn get_crypto_pair_info(&self, key: &str) -> Option<CryptoPairInfo> {
        self.crypto_pairs.get(key).map(|v| v.clone())
    }

    fn crypto_trade_config_key(broker_id: u32, symbol: &str, exchange: &str) -> String {
        format!(
            "{broker_id}:{}:{}",
            symbol.trim().to_ascii_uppercase(),
            exchange.trim().to_ascii_uppercase()
        )
    }

    /// 用 backend CMD20102 拉回的配置替换某个 broker 的 crypto trade config。
    pub fn set_crypto_trade_configs_for_broker(
        &self,
        broker_id: u32,
        configs: Vec<CryptoTradeConfig>,
    ) {
        let prefix = format!("{broker_id}:");
        self.crypto_trade_configs
            .retain(|key, _| !key.starts_with(&prefix));
        for config in configs {
            if config.symbol.trim().is_empty() || config.exchange.trim().is_empty() {
                continue;
            }
            let key = Self::crypto_trade_config_key(broker_id, &config.symbol, &config.exchange);
            self.crypto_trade_configs.insert(key, config);
        }
    }

    /// 查询某个 crypto symbol 的交易配置。
    pub fn get_crypto_trade_config(
        &self,
        broker_id: u32,
        symbol: &str,
        exchange: &str,
    ) -> Option<CryptoTradeConfig> {
        let key = Self::crypto_trade_config_key(broker_id, symbol, exchange);
        self.crypto_trade_configs.get(&key).map(|v| v.clone())
    }

    /// v1.4.106 codex 1148 F9 (P3): 统一写入口 — 部分静态行 (`OnDemandBasic`
    /// source)。同步维护 `securities` + `id_to_key`, **不动** `owner_to_warrants`
    /// (因为 OnDemandBasic 不含 `warrnt_stock_owner` 字段, value 必为 0)。
    ///
    /// `info.source` 必须是 `OnDemandBasic` (debug_assert)。
    pub fn upsert_basic_security_info(&self, key: &str, info: CachedSecurityInfo) {
        debug_assert!(
            !info.source.is_complete(),
            "upsert_basic_security_info called with complete source ({:?}); use upsert_full",
            info.source
        );
        debug_assert_eq!(
            info.warrnt_stock_owner, 0,
            "OnDemandBasic must have warrnt_stock_owner=0 (caller didn't query the field)"
        );
        let _identity_write = match self.zero_id_repair.lock() {
            Ok(guard) => guard,
            Err(_) => return,
        };
        self.upsert_basic_security_info_locked(key, info);
    }

    /// Publish an on-demand row only into the exact stock-id view.
    ///
    /// C++ `SetOptInfo` always updates its option ID map for a valid option ID,
    /// but updates the code map only when the option code is non-empty. This
    /// boundary preserves that distinction without manufacturing a public
    /// empty-code key in the Rust cache.
    pub fn upsert_basic_security_info_by_stock_id_only(&self, info: CachedSecurityInfo) {
        debug_assert!(
            !info.source.is_complete(),
            "ID-only on-demand rows must not replace complete stock-list rows"
        );
        if info.stock_id == 0 {
            return;
        }
        let _identity_write = match self.zero_id_repair.lock() {
            Ok(guard) => guard,
            Err(_) => return,
        };
        if self
            .securities_by_stock_id
            .get(&info.stock_id)
            .is_some_and(|existing| existing.is_complete() || !existing.code.is_empty())
        {
            return;
        }
        self.securities_by_stock_id
            .insert(info.stock_id, Arc::new(info));
    }

    fn upsert_basic_security_info_locked(&self, key: &str, info: CachedSecurityInfo) {
        // 不维护 owner_to_warrants (basic 没这个字段).
        // 但仍需检查既有完整行的 owner 是否会被 basic 错误覆盖.
        // 策略: 如果 key 已有 StockListFull / Bootstrap 行, 不让 basic 行覆盖 (full
        // 数据更完整). 这处理 case "subscribe on-demand 后, stock-list sync 来时
        // 应该 prevail; 反过来不行".
        if let Some(existing) = self.securities.get(key)
            && existing.is_complete()
        {
            tracing::debug!(
                key,
                "upsert_basic_security_info skipped: existing complete row prevails"
            );
            return;
        }
        if info.stock_id > 0
            && self
                .securities_by_stock_id
                .get(&info.stock_id)
                .is_some_and(|existing| existing.is_complete())
        {
            tracing::debug!(
                key,
                stock_id = info.stock_id,
                "upsert_basic_security_info skipped: exact complete row prevails"
            );
            return;
        }
        self.upsert_with_owner_index_maintenance(key, info);
    }

    /// Publish one validated CMD20106 on-demand row without allowing the
    /// primary and reverse identity indexes to diverge.
    ///
    /// The caller may update partial metadata only after `BasicPublished`.
    /// Complete rows and zero-id repairs preserve their authoritative metadata.
    pub fn publish_on_demand_security_info(
        &self,
        key: &str,
        refreshed: CachedSecurityInfo,
        crypto_pair: Option<CryptoPairInfo>,
    ) -> OnDemandSecurityPublishOutcome {
        self.publish_on_demand_security_info_with_pair_hook(key, refreshed, crypto_pair, || {})
    }

    pub(super) fn publish_on_demand_security_info_with_pair_hook(
        &self,
        key: &str,
        refreshed: CachedSecurityInfo,
        crypto_pair: Option<CryptoPairInfo>,
        before_pair_publish: impl FnOnce(),
    ) -> OnDemandSecurityPublishOutcome {
        let _identity_write = match self.zero_id_repair.lock() {
            Ok(guard) => guard,
            Err(_) => return OnDemandSecurityPublishOutcome::Rejected,
        };
        if refreshed.stock_id == 0
            || refreshed.source.is_complete()
            || refreshed.warrnt_stock_owner != 0
        {
            return OnDemandSecurityPublishOutcome::Rejected;
        }

        let stock_id = refreshed.stock_id;
        if self
            .id_to_key
            .get(&stock_id)
            .is_some_and(|mapped| mapped.as_str() != key)
        {
            return OnDemandSecurityPublishOutcome::Rejected;
        }

        let existing = self
            .securities
            .get(key)
            .map(|entry| Arc::clone(entry.value()));
        match existing {
            Some(existing) if existing.stock_id == 0 => {
                let stock_id = refreshed.stock_id;
                if self.repair_zero_stock_id_from_basic_locked(key, refreshed, || {}) {
                    if let Some(pair) = crypto_pair {
                        self.upsert_crypto_pair_by_stock_id_locked(stock_id, pair);
                        self.refresh_public_crypto_pair_locked(key);
                    }
                    OnDemandSecurityPublishOutcome::ZeroIdRepaired
                } else {
                    OnDemandSecurityPublishOutcome::Rejected
                }
            }
            Some(existing) if existing.stock_id != stock_id => {
                OnDemandSecurityPublishOutcome::Rejected
            }
            Some(existing) if existing.is_complete() => {
                if self.update_mkt_id_locked(key, refreshed.mkt_id) {
                    OnDemandSecurityPublishOutcome::CompleteMktIdUpdated
                } else {
                    OnDemandSecurityPublishOutcome::Rejected
                }
            }
            Some(_) | None => {
                let stock_id = refreshed.stock_id;
                self.upsert_basic_security_info_locked(key, refreshed);
                before_pair_publish();
                if let Some(pair) = crypto_pair {
                    self.upsert_crypto_pair_by_stock_id_locked(stock_id, pair);
                    self.refresh_public_crypto_pair_locked(key);
                }
                OnDemandSecurityPublishOutcome::BasicPublished
            }
        }
    }

    /// Repair an existing zero-id row from a validated on-demand CMD20106 row.
    ///
    /// This is deliberately narrower than either upsert path: the primary key
    /// must still identify the same public market/code, the existing row must
    /// still have `stock_id == 0`, and the positive stock id must not belong to
    /// another key. Complete stock-list/bootstrap fields remain authoritative;
    /// only their missing identity and backend market id are repaired.
    pub fn repair_zero_stock_id_from_basic(
        &self,
        key: &str,
        refreshed: CachedSecurityInfo,
    ) -> bool {
        self.repair_zero_stock_id_from_basic_with_security_hook(key, refreshed, || {})
    }

    pub(super) fn repair_zero_stock_id_from_basic_with_security_hook(
        &self,
        key: &str,
        refreshed: CachedSecurityInfo,
        after_security_locked: impl FnOnce(),
    ) -> bool {
        // One repair may need to update primary, reverse, owner, and future
        // alias indices as one bounded cache transaction. Serializing this
        // rare path prevents one repair from holding a security shard while
        // another holds the target positive-id shard and scans securities.
        let _repair = match self.zero_id_repair.lock() {
            Ok(guard) => guard,
            Err(_) => return false,
        };

        self.repair_zero_stock_id_from_basic_locked(key, refreshed, after_security_locked)
    }

    fn repair_zero_stock_id_from_basic_locked(
        &self,
        key: &str,
        refreshed: CachedSecurityInfo,
        after_security_locked: impl FnOnce(),
    ) -> bool {
        if refreshed.stock_id == 0
            || refreshed.source.is_complete()
            || refreshed.warrnt_stock_owner != 0
        {
            return false;
        }

        let stock_id = refreshed.stock_id;
        let old_public_keys = self.public_keys_for_stock_id_locked(stock_id);
        if self
            .id_to_key
            .get(&stock_id)
            .is_some_and(|mapped| mapped.as_str() != key)
        {
            return false;
        }

        let Some(preliminary_ref) = self.securities.get(key) else {
            return false;
        };
        let preliminary = Arc::clone(preliminary_ref.value());
        drop(preliminary_ref);
        if preliminary.stock_id != 0
            || preliminary.market != refreshed.market
            || preliminary.code != refreshed.code
        {
            return false;
        }

        // Owner index publication is serialized before taking a mutable
        // security shard. This lets the winner inspect remaining zero-id rows
        // without forming security-shard -> owner-lock cycles with another
        // repair.
        let preliminary_owner = preliminary.warrnt_stock_owner;
        let mut owner_index = if preliminary_owner != 0 {
            match self.owner_to_warrants.write() {
                Ok(index) => Some(index),
                Err(_) => return false,
            }
        } else {
            None
        };

        let Some(mut entry) = self.securities.get_mut(key) else {
            return false;
        };
        after_security_locked();
        let old = Arc::clone(entry.value());
        if old.stock_id != 0
            || old.market != refreshed.market
            || old.code != refreshed.code
            || old.warrnt_stock_owner != preliminary_owner
        {
            return false;
        }

        let mut updated = if old.is_complete() {
            old.as_ref().clone()
        } else {
            refreshed.clone()
        };
        updated.stock_id = stock_id;
        updated.mkt_id = refreshed.mkt_id;

        let old_owner = old.warrnt_stock_owner;
        let new_owner = updated.warrnt_stock_owner;
        debug_assert_eq!(old_owner, preliminary_owner);
        debug_assert!(new_owner == 0 || owner_index.is_some());

        // Hold the positive-id shard reservation through the complete primary
        // and derived-index commit. Competing zero-id repairs and existing
        // upsert writers for the same stock id cannot pass this ownership gate
        // concurrently.
        let stock_id_reservation = match self.id_to_key.entry(stock_id) {
            dashmap::mapref::entry::Entry::Occupied(occupied) => {
                if occupied.get() != key {
                    return false;
                }
                occupied.into_ref()
            }
            dashmap::mapref::entry::Entry::Vacant(vacant) => vacant.insert(key.to_string()),
        };

        self.remove_future_main_link_aliases(key, &old);
        let updated = Arc::new(updated);
        *entry = Arc::clone(&updated);
        self.securities_by_stock_id
            .insert(stock_id, Arc::clone(&updated));
        if let Some(pair) = self.crypto_pairs.get(key).map(|pair| pair.clone()) {
            self.crypto_pairs_by_stock_id.insert(stock_id, pair);
        }
        self.add_future_main_link_aliases(key, &updated);
        drop(entry);

        if let Some(index) = owner_index.as_mut() {
            if old_owner != 0
                && !self.securities.iter().any(|candidate| {
                    candidate.stock_id == 0 && candidate.warrnt_stock_owner == old_owner
                })
                && let Some(warrants) = index.get_mut(&old_owner)
            {
                warrants.remove(&0);
                if warrants.is_empty() {
                    index.remove(&old_owner);
                }
            }
            if new_owner != 0 {
                index.entry(new_owner).or_default().insert(stock_id);
            }
        }

        drop(stock_id_reservation);
        self.id_to_key.remove_if(&0, |_, mapped| mapped == key);
        let new_public_keys = self.public_keys_for_stock_id_locked(stock_id);
        self.reconcile_public_stock_id_keys_locked(stock_id, &old_public_keys, &new_public_keys);
        true
    }

    /// v1.4.106 codex 1148 F9 (P3): 删除 cache row + 同步清三个索引
    /// (`securities`, `id_to_key`, `owner_to_warrants`)。
    ///
    /// 用于 stock-list `delete_flag = true` 场景。
    /// 返 `true` 如果 row 存在并被删除, `false` 如果 stock_id 不在 `id_to_key`。
    pub fn delete_security_info(&self, stock_id: u64) -> bool {
        let _identity_write = match self.zero_id_repair.lock() {
            Ok(guard) => guard,
            Err(_) => return false,
        };
        self.stock_list_publication_generation
            .fetch_add(1, std::sync::atomic::Ordering::Release);
        let old_public_keys = self.public_keys_for_stock_id_locked(stock_id);
        let Some((_, key)) = self.id_to_key.remove(&stock_id) else {
            return false;
        };
        self.option_contracts.remove(&stock_id);
        if stock_id == 0 {
            if let Some((_, old)) = self
                .securities
                .remove_if(&key, |_, info| info.stock_id == 0)
            {
                self.remove_future_main_link_aliases(&key, &old);
                self.remove_owner_relation(&old);
            }
            self.crypto_pairs.remove(&key);
            self.stale_mkt_ids.remove(&key);
            return true;
        }

        self.security_request_aliases.remove(&stock_id);

        if let Some((_, old)) = self.securities_by_stock_id.remove(&stock_id) {
            self.remove_owner_relation(&old);
            self.remove_future_main_link_aliases_if_unreferenced(&key, &old);
        }
        self.crypto_pairs_by_stock_id.remove(&stock_id);
        self.reconcile_public_stock_id_keys_locked(stock_id, &old_public_keys, &HashSet::new());
        for public_key in old_public_keys {
            self.refresh_public_security_locked(&public_key);
        }

        // F6: 该 stock_id 自己也可能是某 owner — 清掉它作为 owner 的 entry
        if let Ok(mut map) = self.owner_to_warrants.write() {
            map.remove(&stock_id);
        }
        true
    }

    /// 内部 helper: F9 unified upsert with owner-index maintenance (F6).
    fn upsert_with_owner_index_maintenance(&self, key: &str, info: CachedSecurityInfo) {
        let stock_id = info.stock_id;
        if stock_id == 0 {
            self.upsert_zero_id_security_locked(key, info);
            return;
        }

        let old_public_keys = self.public_keys_for_stock_id_locked(stock_id);
        let old_key = self.id_to_key.get(&stock_id).map(|mapped| mapped.clone());
        let old_exact = self
            .securities_by_stock_id
            .remove(&stock_id)
            .map(|(_, old)| old);
        self.id_to_key.remove(&stock_id);

        if let (Some(old_key), Some(old)) = (old_key.as_deref(), old_exact.as_ref())
            && !old.is_complete()
            && old_key != key
        {
            self.security_request_aliases
                .entry(stock_id)
                .or_default()
                .insert(old_key.to_string());
        }
        self.remove_security_request_alias_locked(stock_id, key);

        if let (Some(old_key), Some(old)) = (old_key.as_deref(), old_exact.as_ref()) {
            self.remove_owner_relation(old);
            self.remove_future_main_link_aliases_if_unreferenced(old_key, old);
        }

        self.remove_zero_id_public_row_locked(key);

        let info = Arc::new(info);
        self.securities_by_stock_id
            .insert(stock_id, Arc::clone(&info));
        self.id_to_key.insert(stock_id, key.to_string());
        self.add_owner_relation(&info);
        self.add_future_main_link_aliases(key, &info);

        let new_public_keys = self.public_keys_for_stock_id_locked(stock_id);
        self.reconcile_public_stock_id_keys_locked(stock_id, &old_public_keys, &new_public_keys);
        let mut public_keys_to_refresh = old_public_keys;
        public_keys_to_refresh.extend(new_public_keys);
        for public_key in public_keys_to_refresh {
            self.refresh_public_security_locked(&public_key);
        }
    }

    fn upsert_zero_id_security_locked(&self, key: &str, info: CachedSecurityInfo) {
        if self.public_security_candidate_locked(key).is_some() {
            return;
        }
        self.remove_zero_id_public_row_locked(key);
        let info = Arc::new(info);
        self.securities.insert(key.to_string(), Arc::clone(&info));
        self.id_to_key.insert(0, key.to_string());
        self.add_owner_relation(&info);
        self.add_future_main_link_aliases(key, &info);
    }

    fn remove_zero_id_public_row_locked(&self, key: &str) {
        let Some((_, old)) = self.securities.remove_if(key, |_, info| info.stock_id == 0) else {
            return;
        };
        self.id_to_key.remove_if(&0, |_, mapped| mapped == key);
        self.remove_future_main_link_aliases(key, &old);
        self.remove_owner_relation(&old);
    }

    /// C++ `security.id` is an INTEGER PRIMARY KEY and `index_code` yields
    /// equal-code rows in rowid order. After `SearchSecByCode` filters
    /// `no_search`, `GetStockID` takes the first result, hence the lowest
    /// searchable stock id is the stable public identity.
    fn public_security_candidate_locked(&self, key: &str) -> Option<Arc<CachedSecurityInfo>> {
        let stock_ids = self.public_stock_ids_by_key.get(key)?;
        stock_ids.iter().find_map(|stock_id| {
            #[cfg(test)]
            self.public_security_candidate_inspections
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            self.securities_by_stock_id
                .get(stock_id)
                .map(|info| Arc::clone(info.value()))
                .filter(|info| !info.no_search)
        })
    }

    fn public_keys_for_stock_id_locked(&self, stock_id: u64) -> HashSet<String> {
        if stock_id == 0 {
            return HashSet::new();
        }
        let mut keys = self.security_request_alias_keys_locked(stock_id);
        if let Some(canonical) = self.id_to_key.get(&stock_id) {
            keys.insert(canonical.clone());
        }
        keys
    }

    fn reconcile_public_stock_id_keys_locked(
        &self,
        stock_id: u64,
        old_keys: &HashSet<String>,
        new_keys: &HashSet<String>,
    ) {
        debug_assert!(stock_id > 0);
        for key in old_keys.difference(new_keys) {
            let remove_entry =
                if let Some(mut stock_ids) = self.public_stock_ids_by_key.get_mut(key) {
                    stock_ids.remove(&stock_id);
                    stock_ids.is_empty()
                } else {
                    false
                };
            if remove_entry {
                self.public_stock_ids_by_key.remove(key);
            }
        }
        for key in new_keys.difference(old_keys) {
            self.public_stock_ids_by_key
                .entry(key.clone())
                .or_default()
                .insert(stock_id);
        }
    }

    #[cfg(test)]
    pub(super) fn public_identity_index_matches_sources(&self) -> bool {
        let mut expected =
            std::collections::HashMap::<String, std::collections::BTreeSet<u64>>::new();
        for mapped in &self.id_to_key {
            if *mapped.key() > 0 {
                expected
                    .entry(mapped.value().clone())
                    .or_default()
                    .insert(*mapped.key());
            }
        }
        for aliases in &self.security_request_aliases {
            for alias in aliases.value() {
                expected
                    .entry(alias.clone())
                    .or_default()
                    .insert(*aliases.key());
            }
        }
        let actual = self
            .public_stock_ids_by_key
            .iter()
            .map(|entry| (entry.key().clone(), entry.value().clone()))
            .collect::<std::collections::HashMap<_, _>>();
        actual == expected
    }

    #[cfg(test)]
    pub(super) fn reset_public_security_candidate_inspections(&self) {
        self.public_security_candidate_inspections
            .store(0, std::sync::atomic::Ordering::Relaxed);
    }

    #[cfg(test)]
    pub(super) fn public_security_candidate_inspections(&self) -> u64 {
        self.public_security_candidate_inspections
            .load(std::sync::atomic::Ordering::Relaxed)
    }

    fn security_request_alias_keys_locked(&self, stock_id: u64) -> HashSet<String> {
        self.security_request_aliases
            .get(&stock_id)
            .map(|aliases| aliases.clone())
            .unwrap_or_default()
    }

    fn remove_security_request_alias_locked(&self, stock_id: u64, key: &str) {
        let remove_entry =
            if let Some(mut aliases) = self.security_request_aliases.get_mut(&stock_id) {
                aliases.remove(key);
                aliases.is_empty()
            } else {
                false
            };
        if remove_entry {
            self.security_request_aliases.remove(&stock_id);
        }
    }

    pub(super) fn refresh_public_keys_for_stock_id_locked(&self, stock_id: u64) {
        for key in self.public_keys_for_stock_id_locked(stock_id) {
            self.refresh_public_security_locked(&key);
        }
    }

    fn refresh_public_security_locked(&self, key: &str) {
        if let Some(candidate) = self.public_security_candidate_locked(key) {
            if !candidate.needs_mkt_id_refresh() {
                self.stale_mkt_ids.remove(key);
            }
            self.securities.insert(key.to_string(), candidate);
        } else if self
            .securities
            .get(key)
            .is_some_and(|current| current.stock_id > 0)
        {
            self.securities.remove(key);
            self.stale_mkt_ids.remove(key);
        }
        self.refresh_public_crypto_pair_locked(key);
    }
}