futu_backend/

indicator_metadata.rs

use std::sync::Arc;

use bytes::Bytes;
use prost::Message as _;

use futu_core::error::{FutuError, Result};
use futu_domain_qot_indicator::{
    CatalogPushResult, IndicatorInfo, IndicatorInputValue, IndicatorLanguage, IndicatorParam,
    IndicatorPushOp, IndicatorRefreshOwner, IndicatorRefreshTicket, RefreshProgress,
};

use crate::proto_internal::{
    ft_index_info_common as common, ft_python_indicator_info as py_info,
    ft_python_indicator_param as py_param,
};
use crate::{command_runtime::execute_indicator_read, conn::BackendConn};

use futu_command_spec::IndicatorOperation;

#[must_use]
pub fn build_mylang_info_request(uid: u64, start_guid: &str) -> common::FTindexDownloadReq {
    common::FTindexDownloadReq {
        uid: Some(uid),
        guid: Vec::new(),
        last_updated_time: None,
        start_guid: non_empty(start_guid),
    }
}

#[must_use]
pub fn build_mylang_param_request(uid: u64, start_guid: &str) -> common::FtIndexParamDownloadReq {
    common::FtIndexParamDownloadReq {
        uid: Some(uid),
        guid: Vec::new(),
        last_updated_time: None,
        start_guid: non_empty(start_guid),
    }
}

#[must_use]
pub fn build_python_info_request(start_guid: &str) -> py_info::PythonIndicatorInfoDownloadReq {
    py_info::PythonIndicatorInfoDownloadReq {
        start_guid: non_empty(start_guid),
    }
}

#[must_use]
pub fn build_python_param_request(start_guid: &str) -> py_param::PythonIndicatorParamDownloadReq {
    py_param::PythonIndicatorParamDownloadReq {
        guid: Vec::new(),
        start_guid: non_empty(start_guid),
    }
}

#[must_use]
pub fn adapt_mylang_info(mut item: common::IndexInfo) -> Option<IndicatorInfo> {
    // C++ `CIndicatorDataManager::UpdateSvrMyLangInfo` normalizes the data-layer
    // object before parser creation/cache publication
    // (`CIndicatorDataManager.cpp:637-649`,
    // `CIndicatorDataManagerDetail.cpp:285-306`). Both download and push share
    // this adapter, so the typed script and retained field-complete payload see
    // exactly the same normalized representation.
    if let Some(script) = item.index_script.as_mut() {
        script.make_ascii_uppercase();
    }
    for parameter in &mut item.index_func_param {
        if let Some(name) = parameter.name.as_mut() {
            name.make_ascii_uppercase();
        }
    }
    if let Some(short_name) = item
        .title
        .as_mut()
        .and_then(|title| title.short_cut.as_mut())
    {
        short_name.make_ascii_uppercase();
    }
    let raw_payload = item.encode_to_vec();
    let guid = item.guid.clone().filter(|guid| !guid.is_empty())?;
    let title = item.title.as_ref();
    let short_name = title
        .and_then(|title| title.short_cut.as_ref())
        .filter(|name| !name.is_empty())
        .cloned()
        .unwrap_or_else(|| cpp_guid_short_name(&guid));
    Some(IndicatorInfo {
        language: IndicatorLanguage::MyLang,
        guid,
        short_name: Some(short_name),
        // Full-name resolution depends on FTString.id translation and current
        // app language. The sync layer owns neither, so it retains the complete
        // title in raw_payload and leaves projection unresolved.
        full_name: None,
        script: item.index_script,
        revision: item.update_timestamp.or(item.create_timestamp),
        raw_payload,
    })
}

#[must_use]
pub fn adapt_python_info(item: common::PythonIndicatorInfo) -> Option<IndicatorInfo> {
    let raw_payload = item.encode_to_vec();
    let guid = item.guid.clone().filter(|guid| !guid.is_empty())?;
    Some(IndicatorInfo {
        language: IndicatorLanguage::Python,
        guid,
        // The backend schema carries no structured Python display metadata.
        // Task 12 must not infer it by scanning arbitrary script text.
        short_name: None,
        full_name: None,
        script: item.script,
        revision: item.update_timestamp.or(item.version.map(i64::from)),
        raw_payload,
    })
}

#[must_use]
pub fn adapt_mylang_param(item: common::IndexParam) -> Option<IndicatorParam> {
    let raw_payload = item.encode_to_vec();
    let guid = item.guid.clone().filter(|guid| !guid.is_empty())?;
    Some(IndicatorParam {
        language: IndicatorLanguage::MyLang,
        guid,
        input_values: item
            .param_float_val
            .iter()
            .map(|value| IndicatorInputValue::Float(format_float(*value)))
            .collect(),
        output_styles: item
            .color_val
            .iter()
            .map(|value| Some(i64::from(*value)))
            .collect(),
        revision: None,
        raw_payload,
    })
}

#[must_use]
pub fn adapt_python_param(item: common::PythonIndicatorParam) -> Option<IndicatorParam> {
    let raw_payload = item.encode_to_vec();
    let guid = item.guid.clone().filter(|guid| !guid.is_empty())?;
    Some(IndicatorParam {
        language: IndicatorLanguage::Python,
        guid,
        input_values: item.input.iter().map(python_input_value).collect(),
        // Full style fields remain in raw_payload; this typed summary preserves
        // one stable entry per declared output without parsing script text.
        output_styles: item
            .output_line
            .iter()
            .map(|style| style.line_type.map(i64::from))
            .collect(),
        revision: None,
        raw_payload,
    })
}

#[must_use]
pub const fn decode_indicator_push_op(value: i32) -> Option<IndicatorPushOp> {
    match value {
        1 => Some(IndicatorPushOp::Add),
        2 => Some(IndicatorPushOp::Update),
        3 => Some(IndicatorPushOp::Delete),
        _ => None,
    }
}

pub async fn refresh_indicator_metadata(
    backend: &BackendConn,
    cache: &Arc<futu_cache::indicator::IndicatorMetadataCache>,
    uid: u64,
    owner: IndicatorRefreshOwner,
    owner_is_current: &(impl Fn(IndicatorRefreshOwner) -> bool + Sync),
) -> Result<Arc<futu_domain_qot_indicator::IndicatorCatalogSnapshot>> {
    ensure_current_owner(owner, owner_is_current)?;
    let ticket = cache.begin_refresh(owner);
    refresh_indicator_metadata_with_ticket(backend, cache, uid, ticket, owner_is_current).await
}

pub async fn refresh_indicator_metadata_with_ticket(
    backend: &BackendConn,
    cache: &Arc<futu_cache::indicator::IndicatorMetadataCache>,
    uid: u64,
    ticket: IndicatorRefreshTicket,
    owner_is_current: &(impl Fn(IndicatorRefreshOwner) -> bool + Sync),
) -> Result<Arc<futu_domain_qot_indicator::IndicatorCatalogSnapshot>> {
    if let Err(error) = ensure_current_owner(ticket.owner(), owner_is_current) {
        cache.fail_stream(ticket, IndicatorLanguage::MyLang, true);
        return Err(error);
    }
    let result = tokio::try_join!(
        download_mylang_info(backend, cache, ticket, uid, owner_is_current),
        download_mylang_param(backend, cache, ticket, uid, owner_is_current),
        download_python_info(backend, cache, ticket, owner_is_current),
        download_python_param(backend, cache, ticket, owner_is_current),
    );
    if let Err(error) = result {
        cache.fail_stream(ticket, IndicatorLanguage::MyLang, true);
        return Err(error);
    }
    let snapshot = cache.snapshot().ok_or_else(|| {
        FutuError::Codec("indicator four-stream refresh finished without publication".into())
    })?;
    if snapshot.owner != ticket.owner() {
        return Err(FutuError::Codec(
            "indicator refresh was superseded before composite publication".into(),
        ));
    }
    Ok(snapshot)
}

pub fn apply_mylang_info_push(
    cache: &futu_cache::indicator::IndicatorMetadataCache,
    owner: IndicatorRefreshOwner,
    body: &[u8],
) -> Result<CatalogPushResult> {
    let push = common::InfoPush::decode(body).map_err(FutuError::Proto)?;
    let op = push_op(push.opr)?;
    let item = push
        .item
        .and_then(adapt_mylang_info)
        .ok_or_else(|| FutuError::Codec("MyLang info push missing guid item".into()))?;
    cache
        .apply_info_push(owner, op, item)
        .map_err(indicator_state_error)
}

pub fn apply_mylang_param_push(
    cache: &futu_cache::indicator::IndicatorMetadataCache,
    owner: IndicatorRefreshOwner,
    body: &[u8],
) -> Result<CatalogPushResult> {
    let push = common::ParamPush::decode(body).map_err(FutuError::Proto)?;
    let op = push_op(push.opr)?;
    let item = push
        .item
        .and_then(adapt_mylang_param)
        .ok_or_else(|| FutuError::Codec("MyLang param push missing guid item".into()))?;
    cache
        .apply_param_push(owner, op, item)
        .map_err(indicator_state_error)
}

pub fn apply_python_info_push(
    cache: &futu_cache::indicator::IndicatorMetadataCache,
    owner: IndicatorRefreshOwner,
    body: &[u8],
) -> Result<CatalogPushResult> {
    let push = py_info::PythonIndicatorInfoPush::decode(body).map_err(FutuError::Proto)?;
    let op = push_op(push.opr)?;
    let item = push
        .item
        .and_then(adapt_python_info)
        .ok_or_else(|| FutuError::Codec("Python info push missing guid item".into()))?;
    cache
        .apply_info_push(owner, op, item)
        .map_err(indicator_state_error)
}

pub fn apply_python_param_push(
    cache: &futu_cache::indicator::IndicatorMetadataCache,
    owner: IndicatorRefreshOwner,
    body: &[u8],
) -> Result<CatalogPushResult> {
    let push = py_param::PythonParamPush::decode(body).map_err(FutuError::Proto)?;
    let op = push_op(push.opr)?;
    let item = push
        .item
        .and_then(adapt_python_param)
        .ok_or_else(|| FutuError::Codec("Python param push missing guid item".into()))?;
    cache
        .apply_param_push(owner, op, item)
        .map_err(indicator_state_error)
}

async fn download_mylang_info(
    backend: &BackendConn,
    cache: &Arc<futu_cache::indicator::IndicatorMetadataCache>,
    ticket: IndicatorRefreshTicket,
    uid: u64,
    owner_is_current: &(impl Fn(IndicatorRefreshOwner) -> bool + Sync),
) -> Result<()> {
    let mut cursor = String::new();
    loop {
        let response = execute_indicator_read(
            backend,
            IndicatorOperation::MyLangInfoDownload,
            Bytes::from(build_mylang_info_request(uid, &cursor).encode_to_vec()),
        )
        .await?;
        let page =
            common::FTindexDownloadRsp::decode(response.body.as_ref()).map_err(FutuError::Proto)?;
        ensure_page_success(page.ret_code, page.ret_msg.as_deref(), "MyLang info")?;
        ensure_current_owner(ticket.owner(), owner_is_current)?;
        let progress = cache
            .apply_info_page(
                ticket,
                IndicatorLanguage::MyLang,
                page.items
                    .into_iter()
                    .filter_map(adapt_mylang_info)
                    .collect(),
                page.next_start_guid.as_deref(),
            )
            .map_err(indicator_state_error)?;
        match progress {
            RefreshProgress::Continue(next) => cursor = next,
            RefreshProgress::Pending | RefreshProgress::Published(_) => return Ok(()),
        }
    }
}

async fn download_mylang_param(
    backend: &BackendConn,
    cache: &Arc<futu_cache::indicator::IndicatorMetadataCache>,
    ticket: IndicatorRefreshTicket,
    uid: u64,
    owner_is_current: &(impl Fn(IndicatorRefreshOwner) -> bool + Sync),
) -> Result<()> {
    let mut cursor = String::new();
    loop {
        let response = execute_indicator_read(
            backend,
            IndicatorOperation::MyLangParamDownload,
            Bytes::from(build_mylang_param_request(uid, &cursor).encode_to_vec()),
        )
        .await?;
        let page = common::FtIndexParamDownloadRsp::decode(response.body.as_ref())
            .map_err(FutuError::Proto)?;
        ensure_page_success(page.ret_code, page.ret_msg.as_deref(), "MyLang param")?;
        ensure_current_owner(ticket.owner(), owner_is_current)?;
        let progress = cache
            .apply_param_page(
                ticket,
                IndicatorLanguage::MyLang,
                page.items
                    .into_iter()
                    .filter_map(adapt_mylang_param)
                    .collect(),
                page.next_start_guid.as_deref(),
            )
            .map_err(indicator_state_error)?;
        match progress {
            RefreshProgress::Continue(next) => cursor = next,
            RefreshProgress::Pending | RefreshProgress::Published(_) => return Ok(()),
        }
    }
}

async fn download_python_info(
    backend: &BackendConn,
    cache: &Arc<futu_cache::indicator::IndicatorMetadataCache>,
    ticket: IndicatorRefreshTicket,
    owner_is_current: &(impl Fn(IndicatorRefreshOwner) -> bool + Sync),
) -> Result<()> {
    let mut cursor = String::new();
    loop {
        let response = execute_indicator_read(
            backend,
            IndicatorOperation::PythonInfoDownload,
            Bytes::from(build_python_info_request(&cursor).encode_to_vec()),
        )
        .await?;
        let page = py_info::PythonIndicatorInfoDownloadRsp::decode(response.body.as_ref())
            .map_err(FutuError::Proto)?;
        ensure_page_success(page.ret_code, page.ret_msg.as_deref(), "Python info")?;
        ensure_current_owner(ticket.owner(), owner_is_current)?;
        let progress = cache
            .apply_info_page(
                ticket,
                IndicatorLanguage::Python,
                page.items
                    .into_iter()
                    .filter_map(adapt_python_info)
                    .collect(),
                page.next_start_guid.as_deref(),
            )
            .map_err(indicator_state_error)?;
        match progress {
            RefreshProgress::Continue(next) => cursor = next,
            RefreshProgress::Pending | RefreshProgress::Published(_) => return Ok(()),
        }
    }
}

async fn download_python_param(
    backend: &BackendConn,
    cache: &Arc<futu_cache::indicator::IndicatorMetadataCache>,
    ticket: IndicatorRefreshTicket,
    owner_is_current: &(impl Fn(IndicatorRefreshOwner) -> bool + Sync),
) -> Result<()> {
    let mut cursor = String::new();
    loop {
        let response = execute_indicator_read(
            backend,
            IndicatorOperation::PythonParamDownload,
            Bytes::from(build_python_param_request(&cursor).encode_to_vec()),
        )
        .await?;
        let page = py_param::PythonIndicatorParamDownloadRsp::decode(response.body.as_ref())
            .map_err(FutuError::Proto)?;
        ensure_page_success(page.ret_code, page.ret_msg.as_deref(), "Python param")?;
        ensure_current_owner(ticket.owner(), owner_is_current)?;
        let progress = cache
            .apply_param_page(
                ticket,
                IndicatorLanguage::Python,
                page.items
                    .into_iter()
                    .filter_map(adapt_python_param)
                    .collect(),
                page.next_start_guid.as_deref(),
            )
            .map_err(indicator_state_error)?;
        match progress {
            RefreshProgress::Continue(next) => cursor = next,
            RefreshProgress::Pending | RefreshProgress::Published(_) => return Ok(()),
        }
    }
}

fn ensure_current_owner(
    owner: IndicatorRefreshOwner,
    owner_is_current: &(impl Fn(IndicatorRefreshOwner) -> bool + Sync),
) -> Result<()> {
    if owner_is_current(owner) {
        Ok(())
    } else {
        Err(FutuError::Codec(
            "indicator refresh belongs to a retired auth/connection generation".into(),
        ))
    }
}

fn ensure_page_success(code: Option<i32>, message: Option<&str>, label: &str) -> Result<()> {
    // Proto2 scalar absence reads as the declared default 0 in C++
    // (`NNBiz_Qot_IndicatorInfo.cpp:68-76`,
    // `NNBiz_Qot_IndicatorParam.cpp:72-80`). This is wire-default parity, not
    // a fabricated business success value.
    match code.unwrap_or(0) {
        0 => Ok(()),
        code => Err(FutuError::ServerError {
            ret_type: code,
            msg: message
                .map(str::to_string)
                .unwrap_or_else(|| format!("{label} backend rejected request")),
        }),
    }
}

fn push_op(value: Option<i32>) -> Result<IndicatorPushOp> {
    value
        .and_then(decode_indicator_push_op)
        .ok_or_else(|| FutuError::Codec("indicator push has invalid operation".into()))
}

fn indicator_state_error(error: futu_domain_qot_indicator::IndicatorApplyError) -> FutuError {
    FutuError::Codec(format!(
        "indicator metadata state rejected update: {error:?}"
    ))
}

fn non_empty(value: &str) -> Option<String> {
    (!value.is_empty()).then(|| value.to_string())
}

fn cpp_guid_short_name(guid: &str) -> String {
    // C++ `std::string::substr(0, 8)` in
    // `CIndicatorDataManagerBuild.cpp:56-69,335-347`. Indicator GUIDs are
    // protocol-defined ASCII hex; retain a character-safe fallback only for an
    // invalid non-ASCII protobuf string rather than panic in diagnostics.
    if guid.is_ascii() {
        guid[..guid.len().min(8)].to_string()
    } else {
        guid.chars().take(8).collect()
    }
}

fn format_float(value: f32) -> String {
    let mut text = format!("{value}");
    if text.contains('.') {
        while text.ends_with('0') {
            text.pop();
        }
        if text.ends_with('.') {
            text.pop();
        }
    }
    text
}

fn python_input_value(value: &common::PythonIndicatorInputParam) -> IndicatorInputValue {
    let Some(raw_kind) = value.r#type else {
        return IndicatorInputValue::Missing {
            declared_type: None,
        };
    };
    let Ok(kind) = common::IndicatorInputType::try_from(raw_kind) else {
        return IndicatorInputValue::UnknownType(raw_kind);
    };
    let missing = || IndicatorInputValue::Missing {
        declared_type: Some(raw_kind),
    };
    match kind {
        common::IndicatorInputType::InputInt => value
            .int_value
            .map(IndicatorInputValue::Int)
            .unwrap_or_else(missing),
        common::IndicatorInputType::InputFloat => value
            .float_value
            .map(|value| IndicatorInputValue::Float(format_float(value)))
            .unwrap_or_else(missing),
        common::IndicatorInputType::InputString => value
            .string_value
            .clone()
            .map(IndicatorInputValue::String)
            .unwrap_or_else(missing),
        common::IndicatorInputType::InputBool => value
            .bool_value
            .map(IndicatorInputValue::Bool)
            .unwrap_or_else(missing),
        common::IndicatorInputType::InputColor => value
            .int_value
            .map(IndicatorInputValue::Color)
            .unwrap_or_else(missing),
        common::IndicatorInputType::InputShape => value
            .int_value
            .map(IndicatorInputValue::Shape)
            .unwrap_or_else(missing),
        common::IndicatorInputType::InputLine => value
            .int_value
            .map(IndicatorInputValue::Line)
            .unwrap_or_else(missing),
    }
}