futu_backend/

command_runtime.rs

//! Compatibility exports and backend transport adapter for `futu-command-runtime`.
//!
//! Pure command execution contracts live in `futu-command-runtime`; this module
//! keeps the stable `futu_backend::command_runtime` import path and wires the
//! production backend connection into the new transport trait.

use async_trait::async_trait;
use bytes::Bytes;
use futu_command_runtime::{
    CommandAuthorizationEvidence, CommandRequest, CommandResponse, CommandTransport,
};
use futu_command_spec::trade::{TradeBackendChannel, trade_account_discovery_command};
use futu_command_spec::{
    AuthRequirement, BackendChannelKind, BrokerDiscoveryOperation, CommandSpecId,
    ConnectionDiscoveryOperation, HeartbeatOperation, backend_extension_command,
    broker_discovery_command, command_spec_by_id, connection_discovery_command, heartbeat_command,
    indicator_command, static_data_read_command, system_read_command, user_cloud_command,
};
pub use futu_command_spec::{
    BackendExtensionOperation, CMD_ALGO_ORDER_LOGS, CMD_BATCH_CLOSE_POSITIONS,
    CMD_BFF_ORDER_CANCEL, CMD_BFF_ORDER_CONFIRM, CMD_BFF_ORDER_DELETE, CMD_BFF_ORDER_DETAIL,
    CMD_BFF_ORDER_NEW_V2, CMD_BFF_ORDER_REPLACE, CMD_CLEAR_FUTURES_POSITIONS, CMD_CUSTOMER_BCAN,
    CMD_CUSTOMER_COMPANY_INFO, CMD_GET_BIZ_GROUP, CMD_GET_BOND_ANSWER_STATE,
    CMD_GET_BOND_POSITION_LIST, CMD_GET_BOND_SINGLE_ASSET, CMD_GET_BOND_TOTAL_ASSET,
    CMD_GET_BOND_TRADE_REMINDER, CMD_GET_CASH_DETAIL, CMD_GET_CASH_LOG, CMD_GET_CN_AH_MARGIN_INFO,
    CMD_GET_HK_MARGIN_INFO, CMD_GET_US_MARGIN_INFO, CMD_POSITION_ACTION_RESULT,
    CMD_POSITION_CORPORATE_ACTION_DERIVATIVE, CMD_POSITION_CORPORATE_ACTION_STOCK,
    CMD_PULL_ACCOUNT_FLAG, CMD_REVERSE_POSITION, CMD_ROLL_POSITION,
    IndicatorOperation as IndicatorCommandOperation, MarketEventOperation, QotReadOperation,
    QotWriteOperation, StaticDataReadOperation, SystemReadOperation,
    TradeAccountDiscoveryOperation, TradeAuthOperation, TradeQueryEnvironment, TradeQueryOperation,
    UserCloudOperation,
    trade::{CryptoTradeOperation, TradeWriteOperation},
};
use futu_core::error::{FutuError, Result as FutuResult};

use crate::conn::{
    BackendConn, BackendWriterAdmission, RequestTimeoutPolicy, WriterAdmissionObserver,
};
use crate::quote_sub::{
    DispatchAccepted, NormalSubscriptionDispatchHooks, QotSubscriptionWriterAdmission,
};

#[path = "command_runtime_policy.rs"]
mod policy;

#[path = "command_runtime_qot.rs"]
mod qot;

pub(crate) use qot::execute_qot_subscription_set_with_dispatch;
pub use qot::{
    execute_market_event, execute_qot_read, execute_qot_read_with_reserved,
    execute_qot_subscription_set,
};

use policy::{BACKEND_COMMAND_TIMEOUT, BackendCommandPolicyPorts};

#[cfg(test)]
#[path = "command_runtime_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "command_runtime_crypto_tests.rs"]
mod crypto_tests;

#[cfg(test)]
#[path = "command_runtime_ordinary_tests.rs"]
mod ordinary_tests;

#[cfg(test)]
pub(crate) use crate::command_runtime_test_support::{
    decode_internal_test_request, decode_internal_test_request_body, encode_internal_test_response,
    execute_internal_test_command,
};

pub use futu_command_runtime::{
    ChannelId, ChannelLifecycleState, ChannelRuntimeAction, CommandExecution,
    CommandExecutionContext, CommandExecutionOutcome, CommandOutcome, CommandRuntime,
    CommandRuntimeAction, CommandRuntimeDecision, CommandRuntimeError, decide_command_outcome,
};

#[derive(Clone)]
pub(crate) struct SubscriptionDispatchContext {
    pub(crate) quote_market_type: u8,
    pub(crate) exact_normal_wire: Vec<u8>,
    pub(crate) hooks: NormalSubscriptionDispatchHooks,
    pub(crate) accepted: std::sync::Arc<parking_lot::Mutex<Option<DispatchAccepted>>>,
}

pub type QotWritePublishTerminal = dyn Fn(u64, &mut dyn FnMut()) -> bool + Send + Sync;

pub struct ChannelBoundBackendTransport<'a> {
    backend: &'a BackendConn,
    channel: BackendChannelKind,
    request_timeout: std::time::Duration,
    timeout_policy: RequestTimeoutPolicy,
    subscription_dispatch: Option<SubscriptionDispatchContext>,
    writer_admission_guard: Option<std::sync::Arc<dyn Fn(u64) -> bool + Send + Sync>>,
    writer_admission_publish_terminal: Option<std::sync::Arc<QotWritePublishTerminal>>,
    writer_admission_accepted: Option<std::sync::Arc<dyn Fn(u64) + Send + Sync>>,
}

impl<'a> ChannelBoundBackendTransport<'a> {
    #[must_use]
    pub fn new(backend: &'a BackendConn, channel: BackendChannelKind) -> Self {
        Self {
            backend,
            channel,
            request_timeout: BACKEND_COMMAND_TIMEOUT,
            timeout_policy: RequestTimeoutPolicy::Disconnect,
            subscription_dispatch: None,
            writer_admission_guard: None,
            writer_admission_publish_terminal: None,
            writer_admission_accepted: None,
        }
    }

    fn keep_connection_on_timeout(mut self) -> Self {
        self.timeout_policy = RequestTimeoutPolicy::KeepConnection;
        self
    }

    fn with_subscription_dispatch(mut self, context: SubscriptionDispatchContext) -> Self {
        self.subscription_dispatch = Some(context);
        self
    }

    fn with_writer_admission_guard(
        mut self,
        guard: std::sync::Arc<dyn Fn(u64) -> bool + Send + Sync>,
    ) -> Self {
        self.writer_admission_guard = Some(guard);
        self
    }

    fn with_writer_admission_accepted(
        mut self,
        accepted: std::sync::Arc<dyn Fn(u64) + Send + Sync>,
    ) -> Self {
        self.writer_admission_accepted = Some(accepted);
        self
    }

    fn with_writer_admission_publish_terminal(
        mut self,
        terminal: std::sync::Arc<QotWritePublishTerminal>,
    ) -> Self {
        self.writer_admission_publish_terminal = Some(terminal);
        self
    }
}

#[async_trait]
impl CommandTransport for ChannelBoundBackendTransport<'_> {
    fn channel_kind(&self) -> Option<BackendChannelKind> {
        Some(self.channel)
    }

    async fn execute(&self, request: CommandRequest) -> FutuResult<CommandResponse> {
        let context = self.subscription_dispatch.as_ref();
        let observer = match request.spec_id {
            CommandSpecId::QotSubscriptionSet => context.map(|context| {
                let quote_market_type = context.quote_market_type;
                let exact_normal_wire = context.exact_normal_wire.clone();
                let hooks = context.hooks.clone();
                let admission_hooks = hooks.clone();
                let accepted = std::sync::Arc::clone(&context.accepted);
                WriterAdmissionObserver {
                    try_admit: std::sync::Arc::new(move |admission: BackendWriterAdmission| {
                        (admission_hooks.try_admit)(QotSubscriptionWriterAdmission {
                            connection_generation: admission.connection_generation,
                            serial_no: admission.serial_no,
                        })
                    }),
                    publish_terminal: None,
                    on_accepted: std::sync::Arc::new(move |admission: BackendWriterAdmission| {
                        let receipt = (hooks.on_accepted)(
                            QotSubscriptionWriterAdmission {
                                connection_generation: admission.connection_generation,
                                serial_no: admission.serial_no,
                            },
                            quote_market_type,
                            exact_normal_wire.clone(),
                        );
                        *accepted.lock() = Some(receipt);
                    }),
                }
            }),
            _ if context.is_some() => {
                return Err(FutuError::Codec(
                    "subscription dispatch context used by non-subscription command".into(),
                ));
            }
            _ if self.writer_admission_guard.is_some()
                || self.writer_admission_publish_terminal.is_some() =>
            {
                let guard = self.writer_admission_guard.clone();
                let terminal = self.writer_admission_publish_terminal.clone();
                let accepted = self.writer_admission_accepted.clone();
                Some(WriterAdmissionObserver {
                    try_admit: std::sync::Arc::new(move |admission| {
                        guard
                            .as_ref()
                            .is_none_or(|guard| guard(admission.connection_generation))
                    }),
                    publish_terminal: terminal.map(|terminal| {
                        std::sync::Arc::new(
                            move |admission: BackendWriterAdmission, publish: &mut dyn FnMut()| {
                                terminal(admission.connection_generation, publish)
                            },
                        )
                            as std::sync::Arc<
                                dyn Fn(BackendWriterAdmission, &mut dyn FnMut()) -> bool
                                    + Send
                                    + Sync,
                            >
                    }),
                    on_accepted: std::sync::Arc::new(move |admission| {
                        if let Some(accepted) = accepted.as_ref() {
                            accepted(admission.connection_generation);
                        }
                    }),
                })
            }
            _ => None,
        };
        execute_on_backend(
            self.backend,
            request,
            self.request_timeout,
            self.timeout_policy,
            observer,
        )
        .await
    }
}

fn runtime_with_transport(
    transport: ChannelBoundBackendTransport<'_>,
    authorization: CommandAuthorizationEvidence,
) -> CommandRuntime<ChannelBoundBackendTransport<'_>, BackendCommandPolicyPorts<'_>> {
    let policy = BackendCommandPolicyPorts::new(transport.backend, authorization);
    CommandRuntime::with_policy(transport, policy)
}

fn login_runtime_with_transport(
    transport: ChannelBoundBackendTransport<'_>,
) -> CommandRuntime<ChannelBoundBackendTransport<'_>, BackendCommandPolicyPorts<'_>> {
    runtime_with_transport(transport, CommandAuthorizationEvidence::LoginSession)
}

fn authorization_for_command(
    spec_id: CommandSpecId,
    trade_cipher: Option<&[u8]>,
) -> FutuResult<CommandAuthorizationEvidence> {
    let spec = command_spec_by_id(spec_id)
        .ok_or_else(|| FutuError::Codec(format!("unknown command spec id: {spec_id:?}")))?;
    match spec.auth {
        AuthRequirement::TradeUnlocked => trade_cipher
            .ok_or_else(|| {
                FutuError::Codec(
                    "command authorization evidence: trade cipher was not provided".to_owned(),
                )
            })
            .and_then(|cipher| {
                CommandAuthorizationEvidence::from_trade_cipher(cipher).map_err(|error| {
                    FutuError::Codec(format!("command authorization evidence: {error}"))
                })
            }),
        AuthRequirement::None | AuthRequirement::Login => {
            Ok(CommandAuthorizationEvidence::LoginSession)
        }
    }
}

pub async fn execute_trade_read(
    backend: &BackendConn,
    operation: TradeQueryOperation,
    environment: TradeQueryEnvironment,
    trade_cipher: Option<&[u8]>,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    let channel = match environment {
        TradeQueryEnvironment::Real => BackendChannelKind::Broker,
        TradeQueryEnvironment::Sim => BackendChannelKind::Platform,
    };
    let spec_id = CommandSpecId::TradeRead {
        operation,
        environment,
    };
    let runtime = runtime_with_transport(
        ChannelBoundBackendTransport::new(backend, channel),
        authorization_for_command(spec_id, trade_cipher)?,
    );
    let execution = runtime
        .execute(CommandExecutionContext::new(spec_id, body))
        .await
        .map_err(|error| FutuError::Codec(format!("trade-read command spec error: {error}")))?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed trade read"
    );
    execution.into_response()
}

pub async fn execute_backend_extension(
    backend: &BackendConn,
    operation: BackendExtensionOperation,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    let spec = backend_extension_command(operation);
    let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
        backend,
        spec.runtime.channel,
    ));
    let execution = runtime
        .execute(CommandExecutionContext::new(
            CommandSpecId::BackendExtension(operation),
            body,
        ))
        .await
        .map_err(|error| {
            FutuError::Codec(format!("backend-extension command spec error: {error}"))
        })?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        evidence_kind = ?report.command_evidence.kind,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed backend extension"
    );
    execution.into_response()
}

pub async fn execute_backend_extension_guarded(
    backend: &BackendConn,
    operation: BackendExtensionOperation,
    body: Bytes,
    guard: std::sync::Arc<dyn Fn(u64) -> bool + Send + Sync>,
) -> FutuResult<CommandResponse> {
    let spec = backend_extension_command(operation);
    let transport = ChannelBoundBackendTransport::new(backend, spec.runtime.channel)
        .with_writer_admission_guard(guard);
    let runtime = login_runtime_with_transport(transport);
    let execution = runtime
        .execute(CommandExecutionContext::new(
            CommandSpecId::BackendExtension(operation),
            body,
        ))
        .await
        .map_err(|error| {
            FutuError::Codec(format!(
                "guarded backend-extension command spec error: {error}"
            ))
        })?;
    execution.into_response()
}

pub async fn execute_trade_auth(
    backend: &BackendConn,
    operation: TradeAuthOperation,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
        backend,
        BackendChannelKind::Broker,
    ));
    let execution = runtime
        .execute(CommandExecutionContext::new(
            CommandSpecId::TradeAuth(operation),
            body,
        ))
        .await
        .map_err(|error| FutuError::Codec(format!("trade-auth command spec error: {error}")))?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed trade auth"
    );
    execution.into_response()
}

pub async fn execute_trade_account_discovery(
    backend: &BackendConn,
    operation: TradeAccountDiscoveryOperation,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    let channel = match trade_account_discovery_command(operation).channel {
        TradeBackendChannel::Broker => BackendChannelKind::Broker,
        TradeBackendChannel::Platform => BackendChannelKind::Platform,
    };
    let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(backend, channel));
    let execution = runtime
        .execute(CommandExecutionContext::new(
            CommandSpecId::TradeAccountDiscovery(operation),
            body,
        ))
        .await
        .map_err(|error| {
            FutuError::Codec(format!(
                "trade-account-discovery command spec error: {error}"
            ))
        })?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed trade account discovery"
    );
    execution.into_response()
}

pub async fn execute_connection_discovery(
    backend: &BackendConn,
    operation: ConnectionDiscoveryOperation,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    let spec = connection_discovery_command(operation);
    let runtime =
        login_runtime_with_transport(ChannelBoundBackendTransport::new(backend, spec.channel));
    let execution = runtime
        .execute(CommandExecutionContext::new(
            CommandSpecId::ConnectionDiscovery(operation),
            body,
        ))
        .await
        .map_err(|error| {
            FutuError::Codec(format!("connection-discovery command spec error: {error}"))
        })?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed connection discovery"
    );
    execution.into_response()
}

pub async fn execute_broker_discovery(
    backend: &BackendConn,
    operation: BrokerDiscoveryOperation,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    if operation == BrokerDiscoveryOperation::ValidBrokerListChangedPush {
        return Err(FutuError::Codec(
            "CMD20177 is push-only and cannot be executed as a request".into(),
        ));
    }
    let spec = broker_discovery_command(operation);
    let mut transport = ChannelBoundBackendTransport::new(backend, spec.runtime.channel);
    if operation == BrokerDiscoveryOperation::ValidBrokerList {
        transport = transport.keep_connection_on_timeout();
    }
    let runtime = login_runtime_with_transport(transport);
    let execution = runtime
        .execute(CommandExecutionContext::new(
            CommandSpecId::BrokerDiscovery(operation),
            body,
        ))
        .await
        .map_err(|error| {
            FutuError::Codec(format!("broker-discovery command spec error: {error}"))
        })?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed broker discovery"
    );
    execution.into_response()
}

pub async fn execute_heartbeat(
    backend: &BackendConn,
    operation: HeartbeatOperation,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    let spec = heartbeat_command(operation);
    let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
        backend,
        spec.runtime.channel,
    ));
    let execution = runtime
        .execute(CommandExecutionContext::new(
            CommandSpecId::Heartbeat(operation),
            body,
        ))
        .await
        .map_err(|error| FutuError::Codec(format!("heartbeat command spec error: {error}")))?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed heartbeat"
    );
    execution.into_response()
}

pub async fn execute_trade_write(
    backend: &BackendConn,
    operation: TradeWriteOperation,
    environment: TradeQueryEnvironment,
    trade_cipher: Option<&[u8]>,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    let channel = match environment {
        TradeQueryEnvironment::Real => BackendChannelKind::Broker,
        TradeQueryEnvironment::Sim => BackendChannelKind::Platform,
    };
    let spec_id = CommandSpecId::TradeWrite {
        operation,
        environment,
    };
    let runtime = runtime_with_transport(
        ChannelBoundBackendTransport::new(backend, channel),
        authorization_for_command(spec_id, trade_cipher)?,
    );
    let execution = runtime
        .execute(CommandExecutionContext::new(spec_id, body))
        .await
        .map_err(|error| FutuError::Codec(format!("trade-write command spec error: {error}")))?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed trade write"
    );
    execution.into_response()
}

pub async fn execute_crypto_trade_command(
    backend: &BackendConn,
    operation: CryptoTradeOperation,
    trade_cipher: Option<&[u8]>,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    let spec_id = CommandSpecId::CryptoTrade(operation);
    let runtime = runtime_with_transport(
        ChannelBoundBackendTransport::new(backend, BackendChannelKind::Broker),
        authorization_for_command(spec_id, trade_cipher)?,
    );
    let execution = runtime
        .execute(CommandExecutionContext::new(spec_id, body))
        .await
        .map_err(|error| FutuError::Codec(format!("crypto-trade command spec error: {error}")))?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed crypto trade command"
    );
    execution.into_response()
}

pub async fn execute_system_read(
    backend: &BackendConn,
    operation: SystemReadOperation,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    let spec = system_read_command(operation);
    let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
        backend,
        spec.runtime.channel,
    ));
    let execution = runtime
        .execute(CommandExecutionContext::new(
            CommandSpecId::SystemRead(operation),
            body,
        ))
        .await
        .map_err(|error| FutuError::Codec(format!("system-read command spec error: {error}")))?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        evidence_kind = ?report.command_evidence.kind,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed system read"
    );
    execution.into_response()
}

pub async fn execute_indicator_read(
    backend: &BackendConn,
    operation: IndicatorCommandOperation,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    let spec = indicator_command(operation);
    let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
        backend,
        spec.runtime.channel,
    ));
    let execution = runtime
        .execute(CommandExecutionContext::new(
            CommandSpecId::Indicator(operation),
            body,
        ))
        .await
        .map_err(|error| FutuError::Codec(format!("indicator command spec error: {error}")))?;
    execution.into_response()
}

pub async fn execute_static_data_read(
    backend: &BackendConn,
    operation: StaticDataReadOperation,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    let spec = static_data_read_command(operation);
    let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
        backend,
        spec.runtime.channel,
    ));
    let execution = runtime
        .execute(CommandExecutionContext::new(
            CommandSpecId::StaticDataRead(operation),
            body,
        ))
        .await
        .map_err(|error| {
            FutuError::Codec(format!("static-data-read command spec error: {error}"))
        })?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed static-data read"
    );
    execution.into_response()
}

pub async fn execute_user_cloud(
    backend: &BackendConn,
    operation: UserCloudOperation,
    body: Bytes,
) -> FutuResult<CommandResponse> {
    if operation == UserCloudOperation::UpdatePush {
        return Err(FutuError::Codec(
            "user-cloud CMD20175 is push-only and cannot be executed as a request".to_string(),
        ));
    }
    let spec = user_cloud_command(operation);
    if spec.runtime.channel != BackendChannelKind::Platform {
        return Err(FutuError::Codec(format!(
            "user-cloud command {} is bound to {:?}",
            spec.runtime.name, spec.runtime.channel
        )));
    }
    let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
        backend,
        spec.runtime.channel,
    ));
    let execution = runtime
        .execute(CommandExecutionContext::new(
            CommandSpecId::UserCloud(operation),
            body,
        ))
        .await
        .map_err(|error| FutuError::Codec(format!("user-cloud command spec error: {error}")))?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed user cloud command"
    );
    execution.into_response()
}

pub async fn execute_qot_plaintext(
    backend: &BackendConn,
    cmd_id: u16,
    body: Bytes,
    reserved: [u8; 10],
) -> FutuResult<CommandResponse> {
    execute_qot_command(backend, CommandSpecId::QotPlaintext(cmd_id), body, reserved).await
}

pub async fn execute_qot_write(
    backend: &BackendConn,
    operation: QotWriteOperation,
    body: Bytes,
    reserved: [u8; 10],
) -> FutuResult<CommandResponse> {
    let id = CommandSpecId::QotWrite(operation);
    let spec = futu_command_spec::command_spec_by_id(id).ok_or_else(|| {
        FutuError::Codec(format!("qot-write command spec missing for {operation:?}"))
    })?;
    let runtime =
        login_runtime_with_transport(ChannelBoundBackendTransport::new(backend, spec.channel));
    let execution = runtime
        .execute(CommandExecutionContext::new(id, body).with_reserved(reserved))
        .await
        .map_err(|error| FutuError::Codec(format!("qot-write command spec error: {error}")))?;
    execution.into_response()
}

pub async fn execute_qot_write_guarded(
    backend: &BackendConn,
    operation: QotWriteOperation,
    body: Bytes,
    reserved: [u8; 10],
    publish_terminal: std::sync::Arc<QotWritePublishTerminal>,
    accepted: std::sync::Arc<dyn Fn(u64) + Send + Sync>,
) -> FutuResult<CommandResponse> {
    let id = CommandSpecId::QotWrite(operation);
    let spec = futu_command_spec::command_spec_by_id(id).ok_or_else(|| {
        FutuError::Codec(format!("qot-write command spec missing for {operation:?}"))
    })?;
    let transport = ChannelBoundBackendTransport::new(backend, spec.channel)
        .with_writer_admission_publish_terminal(publish_terminal)
        .with_writer_admission_accepted(accepted);
    let runtime = login_runtime_with_transport(transport);
    let execution = runtime
        .execute(CommandExecutionContext::new(id, body).with_reserved(reserved))
        .await
        .map_err(|error| {
            FutuError::Codec(format!("guarded qot-write command spec error: {error}"))
        })?;
    execution.into_response()
}

async fn execute_qot_command(
    backend: &BackendConn,
    spec_id: CommandSpecId,
    body: Bytes,
    reserved: [u8; 10],
) -> FutuResult<CommandResponse> {
    let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
        backend,
        BackendChannelKind::Qot,
    ));
    let execution = runtime
        .execute(CommandExecutionContext::new(spec_id, body).with_reserved(reserved))
        .await
        .map_err(|error| FutuError::Codec(format!("qot command spec error: {error}")))?;
    let report = &execution.report;
    tracing::trace!(
        command = report.command_name,
        cmd_id = report.cmd_id,
        channel = ?report.channel,
        outcome = ?report.outcome,
        response_body_len = report.response_body_len,
        "command runtime executed QOT command"
    );
    execution.into_response()
}

async fn execute_on_backend(
    backend: &BackendConn,
    request: CommandRequest,
    request_timeout: std::time::Duration,
    timeout_policy: RequestTimeoutPolicy,
    observer: Option<WriterAdmissionObserver>,
) -> FutuResult<CommandResponse> {
    let frame = backend
        .request_with_reserved_timeout_policy_observed(
            request.cmd_id,
            request.body.to_vec(),
            request.reserved,
            request_timeout,
            timeout_policy,
            observer,
        )
        .await?;
    Ok(CommandResponse {
        request_serial_no: frame.header.serial_no,
        cmd_id: frame.header.cmd_id,
        body: frame.body,
        ex_head: frame.ex_head,
    })
}