futu_cache/

broker_dictionary.rs

//! Atomic broker-name dictionary and durable last-good snapshot.
//!
//! C++ loads `Broker.dat` before login, then CMD18008 replaces the version,
//! issuer table and broker-code table as one logical snapshot. Readers must
//! never observe those maps at different generations.

use std::collections::{BTreeMap, BTreeSet};
use std::fs::{File, OpenOptions};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;

use arc_swap::ArcSwap;
use serde::{Deserialize, Serialize};

// Rust-owned durable format version, not a backend value. Increment it only
// when the persisted four-part snapshot schema changes; unsupported versions
// are rejected wholesale instead of being guessed or partially migrated.
const SCHEMA_VERSION: u32 = 1;
// CMD18008's C++ decompressed payload is capped at 2 MiB
// (`NNBiz_Qot_Broker.cpp:225-230`). JSON escaping/field labels expand that
// representation, so disk input is bounded at 16 MiB. Revisit only if a valid
// full CMD18008 snapshot demonstrably exceeds this derived ceiling.
const MAX_FILE_BYTES: u64 = 16 * 1024 * 1024;

// Ref: `NNBase_Define_Enum.h` NN_AppLanguage and
// `NNBiz_Qot_TenBuySellBroker.cpp:45-63`.
const APP_LANGUAGE_ZH: i32 = 0;
const APP_LANGUAGE_HK: i32 = 1;
const APP_LANGUAGE_EN: i32 = 2;
const APP_LANGUAGE_JA: i32 = 5;

#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CachedBrokerInfo {
    /// Legacy Chinese abbreviation (`nameAbbr.sc`).
    pub name_zh_cn: String,
    /// Legacy English abbreviation (`nameAbbr.en`).
    pub name_en: String,
    /// Legacy Traditional Chinese abbreviation (`nameAbbr.tc`).
    pub name_tc: String,
    /// Legacy Chinese full name (`name.sc`).
    pub full_name_zh_cn: String,
    /// Legacy English full name (`name.en`).
    pub full_name_en: String,
    /// Legacy Traditional Chinese full name (`name.tc`).
    pub full_name_tc: String,
}

impl CachedBrokerInfo {
    /// C++ `NNBiz_Qot_TenBuySellBroker.cpp:43-74` lookup order:
    /// selected-language abbreviation, EN/TC/SC abbreviations, then the same
    /// selected/full-name fallback sequence. Unknown app languages use ZH.
    #[must_use]
    pub fn display_name(&self, app_lang: i32) -> Option<String> {
        let (selected_abbr, selected_full) = match app_lang {
            APP_LANGUAGE_EN | APP_LANGUAGE_JA => (&self.name_en, &self.full_name_en),
            APP_LANGUAGE_HK => (&self.name_tc, &self.full_name_tc),
            APP_LANGUAGE_ZH => (&self.name_zh_cn, &self.full_name_zh_cn),
            _ => (&self.name_zh_cn, &self.full_name_zh_cn),
        };
        [
            selected_abbr.as_str(),
            self.name_en.as_str(),
            self.name_tc.as_str(),
            self.name_zh_cn.as_str(),
            selected_full.as_str(),
            self.full_name_en.as_str(),
            self.full_name_tc.as_str(),
            self.full_name_zh_cn.as_str(),
        ]
        .into_iter()
        .find(|name| !name.is_empty())
        .map(ToOwned::to_owned)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum BrokerDictionarySnapshotError {
    #[error("broker dictionary version is invalid")]
    InvalidVersion,
    #[error("broker dictionary issuer id is invalid")]
    InvalidIssuer,
    #[error("broker dictionary issuer id is duplicated")]
    DuplicateIssuer,
    #[error("broker dictionary broker code is invalid")]
    InvalidBrokerCode,
    #[error("broker dictionary broker code is duplicated")]
    DuplicateBrokerCode,
    #[error("broker dictionary broker code points to a missing issuer")]
    DanglingBrokerCode,
}

#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct BrokerDictionarySnapshot {
    initialized: bool,
    backend_version: i32,
    issuer_infos: BTreeMap<i64, CachedBrokerInfo>,
    broker_code_to_issuer: BTreeMap<i64, i64>,
}

impl BrokerDictionarySnapshot {
    pub fn try_from_parts(
        backend_version: i32,
        issuer_entries: Vec<(i64, CachedBrokerInfo)>,
        broker_code_entries: Vec<(i64, i64)>,
    ) -> Result<Self, BrokerDictionarySnapshotError> {
        if backend_version < 0 {
            return Err(BrokerDictionarySnapshotError::InvalidVersion);
        }
        let mut issuer_infos = BTreeMap::new();
        for (issuer_id, info) in issuer_entries {
            if issuer_id <= 0 {
                return Err(BrokerDictionarySnapshotError::InvalidIssuer);
            }
            if issuer_infos.insert(issuer_id, info).is_some() {
                return Err(BrokerDictionarySnapshotError::DuplicateIssuer);
            }
        }

        let mut broker_code_to_issuer = BTreeMap::new();
        for (broker_code, issuer_id) in broker_code_entries {
            if broker_code <= 0 {
                return Err(BrokerDictionarySnapshotError::InvalidBrokerCode);
            }
            if !issuer_infos.contains_key(&issuer_id) {
                return Err(BrokerDictionarySnapshotError::DanglingBrokerCode);
            }
            if broker_code_to_issuer
                .insert(broker_code, issuer_id)
                .is_some()
            {
                return Err(BrokerDictionarySnapshotError::DuplicateBrokerCode);
            }
        }
        Ok(Self {
            initialized: true,
            backend_version,
            issuer_infos,
            broker_code_to_issuer,
        })
    }

    #[must_use]
    pub const fn initialized(&self) -> bool {
        self.initialized
    }

    #[must_use]
    pub const fn backend_version(&self) -> i32 {
        self.backend_version
    }

    #[must_use]
    pub const fn issuer_infos(&self) -> &BTreeMap<i64, CachedBrokerInfo> {
        &self.issuer_infos
    }

    #[must_use]
    pub const fn broker_code_to_issuer(&self) -> &BTreeMap<i64, i64> {
        &self.broker_code_to_issuer
    }
}

pub struct BrokerDictionaryCache {
    snapshot: ArcSwap<BrokerDictionarySnapshot>,
}

impl BrokerDictionaryCache {
    #[must_use]
    pub fn new() -> Self {
        Self {
            snapshot: ArcSwap::new(Arc::new(BrokerDictionarySnapshot::default())),
        }
    }

    #[must_use]
    pub fn snapshot(&self) -> Arc<BrokerDictionarySnapshot> {
        self.snapshot.load_full()
    }

    pub fn replace(&self, snapshot: BrokerDictionarySnapshot) {
        debug_assert!(snapshot.initialized());
        self.snapshot.store(Arc::new(snapshot));
    }

    #[must_use]
    pub fn has_last_good(&self) -> bool {
        self.snapshot.load().initialized()
    }

    #[must_use]
    pub fn display_name_for_issuer(&self, issuer_id: i64, app_lang: i32) -> Option<String> {
        self.snapshot
            .load()
            .issuer_infos
            .get(&issuer_id)
            .and_then(|info| info.display_name(app_lang))
    }

    #[must_use]
    pub fn display_name_for_broker_code(&self, broker_code: i64, app_lang: i32) -> Option<String> {
        let snapshot = self.snapshot.load();
        let issuer_id = snapshot.broker_code_to_issuer.get(&broker_code)?;
        snapshot
            .issuer_infos
            .get(issuer_id)
            .and_then(|info| info.display_name(app_lang))
    }

    #[must_use]
    pub fn issuer_for_broker_code(&self, broker_code: i64) -> Option<i64> {
        self.snapshot
            .load()
            .broker_code_to_issuer
            .get(&broker_code)
            .copied()
    }

    #[must_use]
    pub fn contains_key(&self, issuer_id: &i64) -> bool {
        self.snapshot.load().issuer_infos.contains_key(issuer_id)
    }

    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.snapshot.load().issuer_infos.is_empty()
    }

    #[must_use]
    pub fn len(&self) -> usize {
        self.snapshot.load().issuer_infos.len()
    }

    /// Compatibility helper for tests and manually seeded embedded runtimes.
    /// Production CMD18008 publication uses [`Self::replace`].
    pub fn insert(&self, issuer_id: i64, info: CachedBrokerInfo) {
        if issuer_id <= 0 {
            return;
        }
        self.snapshot.rcu(|current| {
            let mut issuer_entries = current
                .issuer_infos
                .iter()
                .map(|(id, info)| (*id, info.clone()))
                .collect::<Vec<_>>();
            issuer_entries.retain(|(id, _)| *id != issuer_id);
            issuer_entries.push((issuer_id, info.clone()));
            let mut broker_code_entries = current
                .broker_code_to_issuer
                .iter()
                .map(|(code, id)| (*code, *id))
                .collect::<Vec<_>>();
            if !current.broker_code_to_issuer.contains_key(&issuer_id) {
                broker_code_entries.push((issuer_id, issuer_id));
            }
            match BrokerDictionarySnapshot::try_from_parts(
                current.backend_version.max(0),
                issuer_entries,
                broker_code_entries,
            ) {
                Ok(snapshot) => Arc::new(snapshot),
                Err(_) => {
                    tracing::error!(
                        error_class = "snapshot_invariant",
                        "broker dictionary compatibility insert preserved last-good snapshot"
                    );
                    Arc::clone(current)
                }
            }
        });
    }
}

impl Default for BrokerDictionaryCache {
    fn default() -> Self {
        Self::new()
    }
}

#[derive(Debug, thiserror::Error)]
pub enum BrokerDictionaryStoreError {
    #[error("broker dictionary store I/O failed: {0}")]
    Io(#[from] std::io::Error),
    #[error("broker dictionary store JSON failed: {0}")]
    Json(#[from] serde_json::Error),
    #[error("broker dictionary store file is too large")]
    TooLarge,
    #[error("broker dictionary store schema is unsupported")]
    UnsupportedSchema,
    #[error("broker dictionary snapshot is uninitialized")]
    Uninitialized,
    #[error("broker dictionary snapshot is invalid: {0}")]
    InvalidSnapshot(#[from] BrokerDictionarySnapshotError),
}

impl BrokerDictionaryStoreError {
    #[must_use]
    pub const fn class(&self) -> &'static str {
        match self {
            Self::Io(_) => "io",
            Self::Json(_) => "json",
            Self::TooLarge => "too_large",
            Self::UnsupportedSchema => "unsupported_schema",
            Self::Uninitialized => "uninitialized",
            Self::InvalidSnapshot(_) => "invalid_snapshot",
        }
    }
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct PersistedBrokerDictionary {
    schema_version: u32,
    backend_version: i32,
    issuer_infos: Vec<PersistedIssuerInfo>,
    broker_code_to_issuer: Vec<PersistedBrokerCode>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct PersistedIssuerInfo {
    issuer_id: i64,
    info: CachedBrokerInfo,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct PersistedBrokerCode {
    broker_code: i64,
    issuer_id: i64,
}

#[derive(Debug, Clone)]
pub struct BrokerDictionaryStore {
    path: PathBuf,
}

impl BrokerDictionaryStore {
    #[must_use]
    pub fn new(path: impl Into<PathBuf>) -> Self {
        Self { path: path.into() }
    }

    #[must_use]
    pub fn path(&self) -> &Path {
        &self.path
    }

    pub fn load(&self) -> Result<Option<BrokerDictionarySnapshot>, BrokerDictionaryStoreError> {
        let mut file = match OpenOptions::new().read(true).open(&self.path) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error.into()),
        };
        let metadata = file.metadata()?;
        if metadata.len() > MAX_FILE_BYTES {
            return Err(BrokerDictionaryStoreError::TooLarge);
        }
        let mut bytes = Vec::with_capacity(metadata.len() as usize);
        file.read_to_end(&mut bytes)?;
        let persisted: PersistedBrokerDictionary = serde_json::from_slice(&bytes)?;
        if persisted.schema_version != SCHEMA_VERSION {
            return Err(BrokerDictionaryStoreError::UnsupportedSchema);
        }
        let issuer_entries = persisted
            .issuer_infos
            .into_iter()
            .map(|entry| (entry.issuer_id, entry.info))
            .collect();
        let broker_code_entries = persisted
            .broker_code_to_issuer
            .into_iter()
            .map(|entry| (entry.broker_code, entry.issuer_id))
            .collect();
        Ok(Some(BrokerDictionarySnapshot::try_from_parts(
            persisted.backend_version,
            issuer_entries,
            broker_code_entries,
        )?))
    }

    pub fn save(
        &self,
        snapshot: &BrokerDictionarySnapshot,
    ) -> Result<(), BrokerDictionaryStoreError> {
        self.prepare_save(snapshot)?.commit()
    }

    /// Serialize and durably write a unique private temp file without changing
    /// the visible last-good path. Callers may perform slow preparation before
    /// acquiring their publication fence, then use
    /// [`PreparedBrokerDictionarySave::commit`] for the terminal rename.
    pub fn prepare_save(
        &self,
        snapshot: &BrokerDictionarySnapshot,
    ) -> Result<PreparedBrokerDictionarySave, BrokerDictionaryStoreError> {
        self.prepare_save_inner(snapshot, || {})
    }

    /// Same two-phase preparation with an observer invoked after the private
    /// temp is opened and permissioned, immediately before payload write/fsync.
    /// Runtime barrier tests use this exact I/O boundary so moving preparation
    /// under a route lock becomes observable.
    #[cfg(any(test, feature = "test-util"))]
    pub fn prepare_save_with_hook(
        &self,
        snapshot: &BrokerDictionarySnapshot,
        during_write: impl FnOnce(),
    ) -> Result<PreparedBrokerDictionarySave, BrokerDictionaryStoreError> {
        self.prepare_save_inner(snapshot, during_write)
    }

    fn prepare_save_inner(
        &self,
        snapshot: &BrokerDictionarySnapshot,
        during_write: impl FnOnce(),
    ) -> Result<PreparedBrokerDictionarySave, BrokerDictionaryStoreError> {
        if !snapshot.initialized() {
            return Err(BrokerDictionaryStoreError::Uninitialized);
        }
        let explicit_parent = self
            .path
            .parent()
            .filter(|path| !path.as_os_str().is_empty());
        let parent = explicit_parent.unwrap_or_else(|| Path::new("."));
        if let Some(explicit_parent) = explicit_parent {
            std::fs::create_dir_all(explicit_parent)?;
        }
        let persisted = PersistedBrokerDictionary {
            schema_version: SCHEMA_VERSION,
            backend_version: snapshot.backend_version,
            issuer_infos: snapshot
                .issuer_infos
                .iter()
                .map(|(issuer_id, info)| PersistedIssuerInfo {
                    issuer_id: *issuer_id,
                    info: info.clone(),
                })
                .collect(),
            broker_code_to_issuer: snapshot
                .broker_code_to_issuer
                .iter()
                .map(|(broker_code, issuer_id)| PersistedBrokerCode {
                    broker_code: *broker_code,
                    issuer_id: *issuer_id,
                })
                .collect(),
        };
        let bytes = serde_json::to_vec(&persisted)?;
        let stem = self
            .path
            .file_name()
            .and_then(|name| name.to_str())
            .unwrap_or("broker-dictionary-v1.json");
        let mut collisions = BTreeSet::new();
        // Sixteen collision-resistant process-local candidates are enough for
        // concurrent daemon saves; exhaustion is surfaced as I/O failure and
        // never permits a non-atomic write. Replace this bound only if the
        // store adopts a cross-process lock or OS-provided unnamed tempfile.
        let mut during_write = Some(during_write);
        for suffix in 0..16_u8 {
            let temporary = parent.join(format!(".{stem}.{}.{suffix}.tmp", std::process::id()));
            let mut options = OpenOptions::new();
            options.write(true).create_new(true);
            #[cfg(unix)]
            {
                use std::os::unix::fs::OpenOptionsExt as _;
                options.mode(0o600);
            }
            let mut file = match options.open(&temporary) {
                Ok(file) => file,
                Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
                    collisions.insert(suffix);
                    continue;
                }
                Err(error) => return Err(error.into()),
            };
            if let Err(error) = set_file_private(&temporary) {
                let _ = std::fs::remove_file(&temporary);
                return Err(error.into());
            }
            if let Some(during_write) = during_write.take() {
                during_write();
            }
            if let Err(error) = write_prepared(&mut file, &bytes) {
                let _ = std::fs::remove_file(&temporary);
                return Err(error.into());
            }
            return Ok(PreparedBrokerDictionarySave {
                temporary,
                destination: self.path.clone(),
                parent: parent.to_path_buf(),
                committed: false,
            });
        }
        Err(std::io::Error::other(format!(
            "broker dictionary temp names unavailable: {} collisions",
            collisions.len()
        ))
        .into())
    }
}

pub struct PreparedBrokerDictionarySave {
    temporary: PathBuf,
    destination: PathBuf,
    parent: PathBuf,
    committed: bool,
}

impl PreparedBrokerDictionarySave {
    /// Atomically replace the visible snapshot. This terminal step contains no
    /// JSON serialization or payload write. The prepared file is synchronized
    /// before this safe cross-platform rename, and Unix also synchronizes the
    /// containing directory after publication.
    pub fn commit(mut self) -> Result<(), BrokerDictionaryStoreError> {
        atomic_replace(&self.temporary, &self.destination)?;
        self.committed = true;
        set_file_private(&self.destination)?;
        sync_directory(&self.parent)?;
        Ok(())
    }
}

impl Drop for PreparedBrokerDictionarySave {
    fn drop(&mut self) {
        if !self.committed {
            let _ = std::fs::remove_file(&self.temporary);
        }
    }
}

#[must_use]
pub fn default_broker_dictionary_path_from_home(home: &Path) -> PathBuf {
    home.join(".futu-opend-rs")
        .join("broker-dictionary-v1.json")
}

#[must_use]
pub fn default_broker_dictionary_store() -> Option<BrokerDictionaryStore> {
    std::env::var_os("HOME")
        .or_else(|| std::env::var_os("USERPROFILE"))
        .map(PathBuf::from)
        .map(|home| BrokerDictionaryStore::new(default_broker_dictionary_path_from_home(&home)))
}

fn write_prepared(file: &mut File, bytes: &[u8]) -> std::io::Result<()> {
    file.write_all(bytes)?;
    file.sync_all()
}

fn atomic_replace(temporary: &Path, destination: &Path) -> std::io::Result<()> {
    // Rust 1.95's std::fs::rename contract replaces an existing destination on
    // both Unix and Windows. On Windows the standard library owns the
    // MoveFileExW / SetFileInformationByHandle details, so this crate does not
    // need a local unsafe FFI boundary.
    std::fs::rename(temporary, destination)
}

#[cfg(unix)]
fn set_file_private(path: &Path) -> std::io::Result<()> {
    use std::os::unix::fs::PermissionsExt;
    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
}

#[cfg(not(unix))]
fn set_file_private(_path: &Path) -> std::io::Result<()> {
    Ok(())
}

#[cfg(not(windows))]
fn sync_directory(path: &Path) -> std::io::Result<()> {
    File::open(path)?.sync_all()
}

#[cfg(windows)]
fn sync_directory(_path: &Path) -> std::io::Result<()> {
    // The prepared file was synchronized before std::fs::rename; opening a
    // directory as `File` is not portable on Windows.
    Ok(())
}