maxt 0.2.1

//! Upbit spot adapter for Korea, Singapore, Indonesia, and Thailand.

mod parse;
mod private;
mod rest;
mod stream;
mod travel_rule;
mod wallet;

use futures_core::Stream;
use futures_util::StreamExt;
use rust_decimal::Decimal;

use crate::adapter::{Adapter, BoxFuture};
use crate::error::{Error, Result};
use crate::feature::Feature;
use crate::request::{
    CancelOrdersRequest, CandleRequest, DepositAddressRequest, OrderHistoryRequest,
    OrderLookupRequest, OrderRequest, TransferHistoryRequest, TransferLookupRequest,
    WithdrawRequest,
};
use crate::stream::{AccountStream, MarketStream};
use crate::transport::{HttpTransport, WsCommand, WsConnect, WsSession, ws};
use crate::types::{
    AccountEvent, AssetNetwork, Balance, CancelOrdersResult, Candle, Deposit, DepositAddress,
    DepositAddressEntry, Exchange, Market, MarketEvent, MarketInfo, MarketKind, Network, Order,
    OrderBook, OrderRules, Page, Side, StreamConfig, Subscription, Ticker, TimeInForce, Trade,
    TransferDestination, Withdrawal, WithdrawalQuote,
};

pub use travel_rule::{UpbitTravelRuleVasp, UpbitTravelRuleVerification};

/// Selects an Upbit regional deployment.
///
/// Listings, order books, accounts, and credentials are isolated by region.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
#[non_exhaustive]
pub enum UpbitRegion {
    /// Upbit Korea. The default.
    #[default]
    Korea,
    /// Upbit Singapore.
    Singapore,
    /// Upbit Indonesia.
    Indonesia,
    /// Upbit Thailand.
    Thailand,
}

impl UpbitRegion {
    pub(crate) const fn rest_base_url(self) -> &'static str {
        match self {
            Self::Korea => "https://api.upbit.com",
            Self::Singapore => "https://sg-api.upbit.com",
            Self::Indonesia => "https://id-api.upbit.com",
            Self::Thailand => "https://th-api.upbit.com",
        }
    }

    pub(crate) const fn websocket_url(self) -> &'static str {
        match self {
            Self::Korea => "wss://api.upbit.com/websocket/v1",
            Self::Singapore => "wss://sg-api.upbit.com/websocket/v1",
            Self::Indonesia => "wss://id-api.upbit.com/websocket/v1",
            Self::Thailand => "wss://th-api.upbit.com/websocket/v1",
        }
    }
}

/// Warning and caution data for one listing.
///
/// Returned by [`UpbitAdapter::market_events`].
#[derive(Debug, Clone, PartialEq, Eq, Default)]
#[non_exhaustive]
pub struct UpbitMarketEvent {
    /// Whether Upbit marks the listing with an investment warning (`유의 종목`).
    ///
    /// [`Client::markets`](crate::Client::markets) maps this to
    /// [`MarketStatus::Unknown`](crate::MarketStatus::Unknown). The value does
    /// not state whether new orders are currently accepted.
    pub warning: bool,
    /// Active investment-caution (`주의 종목`) criteria, sorted by Upbit's
    /// criterion name.
    ///
    /// These criteria do not change [`MarketStatus`](crate::MarketStatus).
    /// The list is empty outside [`UpbitRegion::Korea`], whose payload is the
    /// only regional payload that includes the criteria.
    pub cautions: Vec<String>,
}

/// One yearly candle returned by Upbit's quotation API.
///
/// This remains provider-specific because the common [`Interval`] type has no
/// yearly variant. `korea_open_time` is present only when Upbit includes its
/// Korea Standard Time wall-clock field in the response.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UpbitYearCandle {
    /// Market that produced the candle.
    pub market: Market,
    /// UTC opening time of the annual window.
    pub open_time: crate::types::Timestamp,
    /// Korea Standard Time opening time when the regional response includes it.
    pub korea_open_time: Option<crate::types::Timestamp>,
    /// Upbit's response timestamp.
    pub timestamp: crate::types::Timestamp,
    /// First trade price in the annual window.
    pub open: Decimal,
    /// Highest trade price in the annual window.
    pub high: Decimal,
    /// Lowest trade price in the annual window.
    pub low: Decimal,
    /// Last trade price in the annual window.
    pub close: Decimal,
    /// Cumulative base-asset volume in the annual window.
    pub volume: Decimal,
    /// Cumulative quote-asset value in the annual window.
    pub quote_volume: Decimal,
    /// First calendar day of Upbit's annual period, preserved as supplied.
    pub first_day_of_period: String,
}

/// Tick-size and supported order-book aggregation policy for one Upbit market.
///
/// Upbit omits `supported_levels` in some regional responses; that is exposed
/// as an empty list rather than inferred from another region.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UpbitOrderBookInstrument {
    /// Market governed by this policy.
    pub market: Market,
    /// Quote currency named by Upbit's response.
    pub quote_currency: String,
    /// Price increment currently applicable to the market's price band.
    ///
    /// Upbit can change this value when the order price moves into another
    /// band, so it is not a market-wide constant.
    pub tick_size: Decimal,
    /// Valid order-book aggregation levels currently published by Upbit.
    pub supported_levels: Vec<Decimal>,
}

/// Upbit가 반환한 한 자산·네트워크의 입금 가능 정보입니다.
///
/// `network`과 `provider_network`은 응답의 `net_type`을 그대로 보존합니다.
/// Upbit는 이 필드를 null로 반환할 수 있으므로, 요청에 사용한 네트워크로
/// 임의 보정하지 않습니다. 이 정보는 실시간 상태가 아니며 몇 분 지연될 수 있습니다.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UpbitDepositInfo {
    /// 대문자로 정규화한 자산 코드입니다.
    pub asset: String,
    /// Upbit가 응답에 포함한 정규화 네트워크입니다.
    pub network: Option<Network>,
    /// Upbit가 응답에 포함한 원본 네트워크 식별자입니다.
    pub provider_network: Option<String>,
    /// 현재 입금 가능 여부입니다.
    pub is_deposit_possible: bool,
    /// 입금이 불가능할 때 Upbit가 제공한 사유입니다.
    pub deposit_impossible_reason: Option<String>,
    /// Upbit가 처리하는 최소 입금 수량입니다.
    pub minimum_deposit_amount: Decimal,
    /// 입금 반영에 필요한 최소 블록 확인 수입니다.
    pub minimum_deposit_confirmations: u64,
    /// 입금 수량에 적용하는 소수 자릿수입니다.
    pub decimal_precision: u64,
}

/// Upbit's ordering when choosing open orders to cancel.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum UpbitOrderDirection {
    /// Cancel the oldest matching orders first.
    Ascending,
    /// Cancel the newest matching orders first.
    Descending,
}

/// The explicit set of Upbit open orders considered for one batch cancellation.
///
/// [`Self::All`] is deliberately a named variant: it selects every eligible
/// market, while Upbit still applies the request count (default 20, maximum
/// 300) to matching open orders.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum UpbitBatchCancelScope {
    /// Every eligible Upbit market; the request count still limits cancellations.
    All,
    /// Eligible orders in markets with one of these quote currencies.
    QuoteCurrencies {
        /// Quote currencies used to select eligible markets.
        values: Vec<String>,
    },
    /// Eligible orders in these explicit Upbit spot markets.
    Pairs {
        /// Upbit spot markets used to select eligible orders.
        values: Vec<Market>,
    },
}

/// Filters for Upbit's conditional batch-cancellation endpoint.
///
/// The endpoint can cancel at most 300 `wait` orders per request. It never
/// cancels `watch` orders. A successful response can still contain failures
/// because matching orders may fill or change state while Upbit processes the
/// request.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UpbitBatchCancelRequest {
    /// The explicit base set of orders to consider.
    pub scope: UpbitBatchCancelScope,
    /// Optional Upbit spot markets to leave untouched.
    pub excluded_pairs: Option<Vec<Market>>,
    /// Optional buy/sell filter. `None` leaves Upbit's `all` default.
    pub side: Option<Side>,
    /// Optional cancellation count. `None` leaves Upbit's default of 20.
    pub count: Option<u32>,
    /// Optional creation-time ordering. `None` leaves Upbit's `desc` default.
    pub order_by: Option<UpbitOrderDirection>,
}

impl UpbitBatchCancelRequest {
    /// Starts a batch cancellation with an explicit scope.
    pub fn new(scope: UpbitBatchCancelScope) -> Self {
        Self {
            scope,
            excluded_pairs: None,
            side: None,
            count: None,
            order_by: None,
        }
    }

    /// Leaves these Upbit spot markets untouched.
    #[must_use]
    pub fn excluded_pairs(mut self, pairs: impl Into<Vec<Market>>) -> Self {
        self.excluded_pairs = Some(pairs.into());
        self
    }

    /// Limits cancellation to one order side.
    #[must_use]
    pub fn side(mut self, side: Side) -> Self {
        self.side = Some(side);
        self
    }

    /// Limits how many matching orders Upbit cancels.
    #[must_use]
    pub fn count(mut self, count: u32) -> Self {
        self.count = Some(count);
        self
    }

    /// Chooses whether Upbit considers oldest or newest orders first.
    #[must_use]
    pub fn order_by(mut self, order_by: UpbitOrderDirection) -> Self {
        self.order_by = Some(order_by);
        self
    }
}

/// Identifies the existing order for Upbit's cancel-and-new operation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum UpbitOrderReference {
    /// Upbit-issued order UUID.
    Uuid(String),
    /// Caller-assigned order identifier.
    Identifier(String),
}

impl UpbitOrderReference {
    /// Uses an Upbit-issued order UUID.
    pub fn uuid(value: impl Into<String>) -> Self {
        Self::Uuid(value.into())
    }

    /// Uses a caller-assigned order identifier.
    pub fn identifier(value: impl Into<String>) -> Self {
        Self::Identifier(value.into())
    }
}

/// New-order volume for Upbit cancel-and-new.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum UpbitOrderVolume {
    /// Explicit base-asset volume.
    Amount(Decimal),
    /// Reuse the previous order's remaining volume.
    RemainOnly,
}

/// Self-match prevention mode for the replacement order.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum UpbitSmpType {
    /// Cancel the maker order when a self-match would occur.
    CancelMaker,
    /// Cancel the taker order when a self-match would occur.
    CancelTaker,
    /// Reduce both orders by the self-matched amount.
    Reduce,
}

/// The replacement order shape accepted by Upbit's cancel-and-new endpoint.
///
/// The endpoint inherits the previous order's market and side. The buy/sell
/// variants make Upbit's direction-dependent `price`/`volume` fields explicit
/// without pretending that the endpoint accepts a new market or side.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum UpbitCancelAndNewOrder {
    /// Limit order. `time_in_force` may be omitted for Upbit's default GTC.
    Limit {
        /// New base-asset volume, or `RemainOnly`.
        volume: UpbitOrderVolume,
        /// New quote price.
        price: Decimal,
        /// Optional IOC, FOK, or post-only policy.
        time_in_force: Option<TimeInForce>,
    },
    /// Market buy (`new_ord_type = "price"`).
    MarketBuy {
        /// Total quote amount to spend.
        price: Decimal,
    },
    /// Market sell (`new_ord_type = "market"`).
    MarketSell {
        /// New base-asset volume, or `RemainOnly`.
        volume: UpbitOrderVolume,
    },
    /// Best-price buy. Upbit requires IOC or FOK.
    BestBuy {
        /// Total quote amount to spend.
        price: Decimal,
        /// Required IOC or FOK policy.
        time_in_force: TimeInForce,
    },
    /// Best-price sell. Upbit requires IOC or FOK.
    BestSell {
        /// New base-asset volume, or `RemainOnly`.
        volume: UpbitOrderVolume,
        /// Required IOC or FOK policy.
        time_in_force: TimeInForce,
    },
}

/// Request for Upbit's single-request cancel-then-create order operation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UpbitCancelAndNewOrderRequest {
    /// Existing order to cancel.
    pub previous_order: UpbitOrderReference,
    /// Replacement order to create after cancellation.
    pub new_order: UpbitCancelAndNewOrder,
    /// Optional new caller-assigned identifier.
    pub new_identifier: Option<String>,
    /// Optional self-match prevention mode for the replacement order.
    pub new_smp_type: Option<UpbitSmpType>,
}

impl UpbitCancelAndNewOrderRequest {
    /// Starts a cancel-and-new request.
    pub fn new(previous_order: UpbitOrderReference, new_order: UpbitCancelAndNewOrder) -> Self {
        Self {
            previous_order,
            new_order,
            new_identifier: None,
            new_smp_type: None,
        }
    }

    /// Assigns the replacement order's client identifier.
    #[must_use]
    pub fn new_identifier(mut self, value: impl Into<String>) -> Self {
        self.new_identifier = Some(value.into());
        self
    }

    /// Selects the replacement order's self-match prevention mode.
    #[must_use]
    pub fn new_smp_type(mut self, value: UpbitSmpType) -> Self {
        self.new_smp_type = Some(value);
        self
    }
}

/// Result returned by Upbit's cancel-and-new endpoint.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UpbitCancelAndNewOrderResult {
    /// The previous order after Upbit processed the request.
    ///
    /// It is usually cancelled, but it can be filled when it completed before
    /// cancellation. Inspect its status instead of assuming cancellation.
    pub previous_order: Order,
    /// UUID of the replacement order, when one was created.
    ///
    /// This is `None` when the old order filled before cancellation completed;
    /// a successful HTTP response alone does not imply replacement creation.
    pub new_order_uuid: Option<String>,
    /// Identifier of the replacement order, when one was requested and created.
    pub new_order_identifier: Option<String>,
}

impl UpbitCancelAndNewOrderResult {
    /// Whether Upbit reported that a replacement order was created.
    pub fn replacement_created(&self) -> bool {
        self.new_order_uuid.is_some()
    }
}

/// Adapter for Upbit spot markets.
///
/// Derivative features return [`Error::Unsupported`](crate::Error::Unsupported).
#[derive(Debug, Clone)]
pub struct UpbitAdapter {
    region: UpbitRegion,
    credentials: Option<UpbitCredentials>,
    /// Cached transport initialization result, reported on first use.
    http: std::result::Result<HttpTransport, Error>,
}

#[derive(Debug, Clone)]
pub(crate) struct UpbitCredentials {
    pub(crate) access_key: String,
    pub(crate) secret_key: String,
}

impl UpbitAdapter {
    /// Creates an unauthenticated adapter for Upbit Korea.
    pub fn new() -> Self {
        Self::with_region(UpbitRegion::Korea)
    }

    /// Creates an unauthenticated adapter for `region`.
    pub fn with_region(region: UpbitRegion) -> Self {
        Self {
            region,
            credentials: None,
            http: HttpTransport::new(region.rest_base_url()),
        }
    }

    /// Adds credentials for account, order, and private-stream calls.
    ///
    /// The key pair must be issued by the adapter's selected region.
    #[must_use]
    pub fn with_credentials(
        mut self,
        access_key: impl Into<String>,
        secret_key: impl Into<String>,
    ) -> Self {
        self.credentials = Some(UpbitCredentials {
            access_key: access_key.into(),
            secret_key: secret_key.into(),
        });
        self
    }

    /// Returns the selected region.
    pub fn region(&self) -> UpbitRegion {
        self.region
    }

    /// Fetches order books for one or more markets in one REST request.
    ///
    /// `depth` is the number of levels per side and must be from 1 through 30.
    /// `None` uses Upbit's 30-level default.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidRequest`](crate::Error::InvalidRequest) for an
    /// empty market list, an invalid depth, a different exchange, or an invalid
    /// asset code. Non-spot markets return
    /// [`Error::Unsupported`](crate::Error::Unsupported). Transport, exchange,
    /// and decoding errors are propagated.
    pub async fn order_books(
        &self,
        markets: &[Market],
        depth: Option<u32>,
    ) -> Result<Vec<OrderBook>> {
        rest::order_books(self.http()?, markets, depth).await
    }

    /// Fetches Upbit Korea order books aggregated at one provider level.
    ///
    /// `level` must be zero or positive. Read [`Self::orderbook_instruments`]
    /// immediately before this call to select a current non-zero value: Upbit
    /// changes supported levels when a market moves between price bands. Global
    /// regional deployments do not accept this parameter.
    pub async fn order_books_at_level(
        &self,
        markets: &[Market],
        level: Decimal,
        depth: Option<u32>,
    ) -> Result<Vec<OrderBook>> {
        if self.region != UpbitRegion::Korea {
            return Err(Error::unsupported(
                Feature::OrderBook,
                Exchange::Upbit.id(),
                "order-book aggregation levels are available only in the Upbit Korea region",
            ));
        }
        rest::order_books_at_level(self.http()?, markets, level, depth).await
    }

    /// Fetches tickers for one or more markets in one REST request.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidRequest`](crate::Error::InvalidRequest) for an
    /// empty market list, a different exchange, or an invalid asset code.
    /// Non-spot markets return [`Error::Unsupported`](crate::Error::Unsupported).
    /// Transport, exchange, and decoding errors are propagated.
    pub async fn tickers(&self, markets: &[Market]) -> Result<Vec<Ticker>> {
        rest::tickers(self.http()?, markets).await
    }

    /// Fetches every ticker in one or more quote-currency markets.
    ///
    /// The input is normalized to uppercase ASCII currency codes. It must name
    /// at least one non-empty code. This is distinct from [`Self::tickers`],
    /// which queries explicit trading pairs.
    pub async fn tickers_by_quote(&self, quote_currencies: &[String]) -> Result<Vec<Ticker>> {
        rest::tickers_by_quote(self.http()?, quote_currencies).await
    }

    /// Fetches Upbit's yearly candles for one market.
    ///
    /// `to` is an optional exclusive ISO-8601 boundary and `count`, when set,
    /// must be from 1 through 200. Results are oldest first. The endpoint does
    /// not use the common [`crate::types::Candle`] model because its annual
    /// interval is unique to Upbit's current public surface.
    pub async fn year_candles(
        &self,
        market: &Market,
        to: Option<crate::types::Timestamp>,
        count: Option<u32>,
    ) -> Result<Vec<UpbitYearCandle>> {
        rest::year_candles(self.http()?, market, to, count).await
    }

    /// Fetches tick-size and order-book aggregation policy for one or more markets.
    ///
    /// The returned tick size and `supported_levels` list are live provider
    /// metadata. A caller must not assume that either remains valid after the
    /// intended price moves into another band.
    pub async fn orderbook_instruments(
        &self,
        markets: &[Market],
    ) -> Result<Vec<UpbitOrderBookInstrument>> {
        rest::orderbook_instruments(self.http()?, markets).await
    }

    /// Fetches warning and caution data for every listed market.
    ///
    /// Caution criteria are empty outside [`UpbitRegion::Korea`].
    ///
    /// # Errors
    ///
    /// Propagates transport, exchange, and decoding errors.
    pub async fn market_events(&self) -> Result<Vec<(Market, UpbitMarketEvent)>> {
        rest::market_events(self.http()?).await
    }

    /// Validates an order without creating it on Upbit.
    ///
    /// Upbit returns a dry-run order object. Its identifier and status do not
    /// represent a live order, so it cannot be queried or cancelled.
    pub async fn test_order(&self, request: &OrderRequest) -> Result<Order> {
        private::test_order(self.credentials()?, self.http()?, request).await
    }

    /// 한 자산·네트워크의 Upbit 입금 가능 정보를 조회합니다.
    ///
    /// Upbit의 응답은 실시간 서비스 상태를 보장하지 않으며 몇 분 지연될 수 있습니다.
    pub async fn deposit_info(&self, asset: &str, network: &Network) -> Result<UpbitDepositInfo> {
        wallet::deposit_info(self.credentials()?, self.http()?, asset, network).await
    }

    /// Lists VASPs supported by Upbit's Travel Rule service.
    ///
    /// This read requires the API key's `View Deposits` permission and is
    /// available only when this adapter targets [`UpbitRegion::Korea`] or
    /// [`UpbitRegion::Singapore`].
    pub async fn travel_rule_vasps(&self) -> Result<Vec<UpbitTravelRuleVasp>> {
        travel_rule::ensure_supported_region(self.region)?;
        travel_rule::vasps(self.region, self.credentials()?, self.http()?).await
    }

    /// Requests Travel Rule account-owner verification by deposit UUID.
    ///
    /// This is a financial write requiring the API key's `Deposit` permission.
    /// Upbit permits at most one request for the same deposit every 10 minutes;
    /// that repeat restriction is enforced by Upbit, not tracked client-side.
    /// The endpoint is available only in [`UpbitRegion::Korea`] or
    /// [`UpbitRegion::Singapore`].
    pub async fn verify_travel_rule_by_uuid(
        &self,
        deposit_uuid: &str,
        vasp_uuid: &str,
    ) -> Result<UpbitTravelRuleVerification> {
        travel_rule::ensure_supported_region(self.region)?;
        travel_rule::verify_by_uuid(
            self.region,
            self.credentials()?,
            self.http()?,
            deposit_uuid,
            vasp_uuid,
        )
        .await
    }

    /// Requests Travel Rule account-owner verification by deposit transaction ID.
    ///
    /// This is a financial write requiring the API key's `Deposit` permission.
    /// Upbit permits at most one request for the same deposit every 10 minutes;
    /// that repeat restriction is enforced by Upbit, not tracked client-side.
    /// The endpoint is available only in [`UpbitRegion::Korea`] or
    /// [`UpbitRegion::Singapore`].
    pub async fn verify_travel_rule_by_txid(
        &self,
        txid: &str,
        vasp_uuid: &str,
        currency: &str,
        net_type: &str,
    ) -> Result<UpbitTravelRuleVerification> {
        travel_rule::ensure_supported_region(self.region)?;
        travel_rule::verify_by_txid(
            self.region,
            self.credentials()?,
            self.http()?,
            txid,
            vasp_uuid,
            currency,
            net_type,
        )
        .await
    }

    /// Cancels matching Upbit `wait` orders in one conditional request.
    ///
    /// This is a financial write. The returned value separates orders Upbit
    /// cancelled from orders that changed state before cancellation completed.
    pub async fn batch_cancel_open_orders(
        &self,
        request: &UpbitBatchCancelRequest,
    ) -> Result<CancelOrdersResult> {
        private::batch_cancel_open_orders(self.credentials()?, self.http()?, request).await
    }

    /// Cancels one existing order and creates its replacement in one request.
    ///
    /// Upbit keeps the previous order's market and side. A `201` response can
    /// still report no replacement UUID when the previous order filled before
    /// cancellation completed; inspect [`UpbitCancelAndNewOrderResult::replacement_created`]
    /// instead of treating HTTP success as an atomic replacement guarantee.
    pub async fn cancel_and_new_order(
        &self,
        request: &UpbitCancelAndNewOrderRequest,
    ) -> Result<UpbitCancelAndNewOrderResult> {
        private::cancel_and_new_order(self.credentials()?, self.http()?, request).await
    }

    pub(crate) fn is_authenticated(&self) -> bool {
        self.credentials.is_some()
    }

    fn http(&self) -> Result<&HttpTransport> {
        self.http.as_ref().map_err(Clone::clone)
    }

    fn credentials(&self) -> Result<&UpbitCredentials> {
        self.credentials.as_ref().ok_or_else(|| {
            Error::auth(
                "this Upbit adapter has no credentials; add them with \
                 `UpbitAdapter::with_credentials`",
            )
        })
    }

    fn validate_withdrawal_destination(&self, request: &WithdrawRequest) -> Result<()> {
        if self.region == UpbitRegion::Indonesia
            && !matches!(
                &request.destination,
                TransferDestination::Exchange(destination)
                    if destination.exchange == Exchange::Upbit
            )
        {
            return Err(Error::unsupported(
                Feature::Withdrawals,
                "upbit",
                "Upbit Indonesia external withdrawals require beneficiary fields that are not yet represented by the common withdrawal request",
            ));
        }
        Ok(())
    }
}

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

impl Adapter for UpbitAdapter {
    fn exchange(&self) -> Exchange {
        Exchange::Upbit
    }

    fn supports(&self, feature: Feature) -> bool {
        if feature.is_derivatives_only() {
            return false;
        }
        if feature == Feature::TravelRule {
            return matches!(self.region, UpbitRegion::Korea | UpbitRegion::Singapore)
                && self.is_authenticated();
        }
        if feature.needs_credentials() {
            return self.is_authenticated();
        }
        true
    }

    fn markets(&self, kind: MarketKind) -> BoxFuture<'_, Result<Vec<MarketInfo>>> {
        Box::pin(async move { rest::markets(self.http()?, kind).await })
    }

    fn trades(&self, market: &Market, limit: Option<u32>) -> BoxFuture<'_, Result<Vec<Trade>>> {
        let market = market.clone();
        Box::pin(async move { rest::trades(self.http()?, &market, limit).await })
    }

    fn order_book(&self, market: &Market, depth: Option<u32>) -> BoxFuture<'_, Result<OrderBook>> {
        let market = market.clone();
        Box::pin(async move {
            let books = self
                .order_books(std::slice::from_ref(&market), depth)
                .await?;
            rest::only(books, &market)
        })
    }

    fn ticker(&self, market: &Market) -> BoxFuture<'_, Result<Ticker>> {
        let market = market.clone();
        Box::pin(async move {
            let tickers = self.tickers(std::slice::from_ref(&market)).await?;
            rest::only(tickers, &market)
        })
    }

    fn candles(&self, request: &CandleRequest) -> BoxFuture<'_, Result<Vec<Candle>>> {
        let request = request.clone();
        Box::pin(async move { rest::candles(self.http()?, &request).await })
    }

    fn subscribe(
        &self,
        subscription: &Subscription,
        config: &StreamConfig,
    ) -> BoxFuture<'_, Result<MarketStream>> {
        let frame = stream::subscribe_frame(subscription, &ticket());
        let url = self.region.websocket_url().to_string();
        let config = config.clone();

        Box::pin(async move {
            let session = ws::connect(
                WsConnect {
                    url,
                    headers: None,
                    subscribe: WsConnect::fixed(vec![frame?]),
                    heartbeat: Some(stream::HEARTBEAT),
                },
                &config,
            )
            .await?;
            let close = session.close_handle();

            // Candle completion state belongs to one WebSocket connection.
            let mut decoder = stream::Decoder::default();

            Ok(MarketStream::new_with_close(
                events(
                    session,
                    move |frame| decoder.decode(frame),
                    MarketEvent::Reconnected,
                ),
                move || async move { close.close().await },
            ))
        })
    }

    fn balances(&self) -> BoxFuture<'_, Result<Vec<Balance>>> {
        Box::pin(async move { private::balances(self.credentials()?, self.http()?).await })
    }

    fn order_rules(&self, market: &Market) -> BoxFuture<'_, Result<OrderRules>> {
        let market = market.clone();
        Box::pin(
            async move { private::order_rules(self.credentials()?, self.http()?, &market).await },
        )
    }

    fn asset_networks(&self, asset: &str) -> BoxFuture<'_, Result<Vec<AssetNetwork>>> {
        let asset = asset.to_string();
        Box::pin(
            async move { wallet::asset_networks(self.credentials()?, self.http()?, &asset).await },
        )
    }

    fn deposit_addresses(&self) -> BoxFuture<'_, Result<Vec<DepositAddressEntry>>> {
        Box::pin(async move { wallet::deposit_addresses(self.credentials()?, self.http()?).await })
    }

    fn deposit_address(
        &self,
        request: &DepositAddressRequest,
    ) -> BoxFuture<'_, Result<DepositAddress>> {
        let request = request.clone();
        Box::pin(async move {
            wallet::deposit_address(self.credentials()?, self.http()?, &request).await
        })
    }

    fn create_deposit_address(
        &self,
        request: &DepositAddressRequest,
    ) -> BoxFuture<'_, Result<DepositAddress>> {
        let request = request.clone();
        Box::pin(async move {
            wallet::create_deposit_address(self.credentials()?, self.http()?, &request).await
        })
    }

    fn prepare_withdrawal(
        &self,
        request: &WithdrawRequest,
    ) -> BoxFuture<'_, Result<WithdrawalQuote>> {
        let request = request.clone();
        Box::pin(async move {
            self.validate_withdrawal_destination(&request)?;
            wallet::prepare_withdrawal(self.credentials()?, self.http()?, &request).await
        })
    }

    fn withdraw(&self, request: &WithdrawRequest) -> BoxFuture<'_, Result<Withdrawal>> {
        let request = request.clone();
        Box::pin(async move {
            self.validate_withdrawal_destination(&request)?;
            wallet::withdraw(self.credentials()?, self.http()?, &request).await
        })
    }

    fn deposit(&self, request: &TransferLookupRequest) -> BoxFuture<'_, Result<Deposit>> {
        let request = request.clone();
        Box::pin(async move { wallet::deposit(self.credentials()?, self.http()?, &request).await })
    }

    fn withdrawal(&self, request: &TransferLookupRequest) -> BoxFuture<'_, Result<Withdrawal>> {
        let request = request.clone();
        Box::pin(
            async move { wallet::withdrawal(self.credentials()?, self.http()?, &request).await },
        )
    }

    fn cancel_withdrawal(&self, withdrawal_id: &str) -> BoxFuture<'_, Result<()>> {
        let withdrawal_id = withdrawal_id.to_owned();
        Box::pin(async move {
            wallet::cancel_withdrawal(self.credentials()?, self.http()?, &withdrawal_id).await
        })
    }

    fn deposits(&self, request: &TransferHistoryRequest) -> BoxFuture<'_, Result<Page<Deposit>>> {
        let request = request.clone();
        Box::pin(async move { wallet::deposits(self.credentials()?, self.http()?, &request).await })
    }

    fn withdrawals(
        &self,
        request: &TransferHistoryRequest,
    ) -> BoxFuture<'_, Result<Page<Withdrawal>>> {
        let request = request.clone();
        Box::pin(
            async move { wallet::withdrawals(self.credentials()?, self.http()?, &request).await },
        )
    }

    fn open_orders(&self, market: Option<&Market>) -> BoxFuture<'_, Result<Vec<Order>>> {
        let market = market.cloned();
        Box::pin(async move {
            private::open_orders(self.credentials()?, self.http()?, market.as_ref()).await
        })
    }

    fn order(&self, market: &Market, order_id: &str) -> BoxFuture<'_, Result<Order>> {
        let market = market.clone();
        let order_id = order_id.to_string();
        Box::pin(async move {
            private::order(self.credentials()?, self.http()?, &market, &order_id).await
        })
    }

    fn order_by_client_id(&self, market: &Market, client_id: &str) -> BoxFuture<'_, Result<Order>> {
        let market = market.clone();
        let client_id = client_id.to_string();
        Box::pin(async move {
            private::order_by_client_id(self.credentials()?, self.http()?, &market, &client_id)
                .await
        })
    }

    fn orders_by_ids(&self, request: &OrderLookupRequest) -> BoxFuture<'_, Result<Vec<Order>>> {
        let request = request.clone();
        Box::pin(async move {
            private::orders_by_ids(self.credentials()?, self.http()?, &request).await
        })
    }

    fn order_history(&self, request: &OrderHistoryRequest) -> BoxFuture<'_, Result<Page<Order>>> {
        let request = request.clone();
        Box::pin(async move {
            private::order_history(self.credentials()?, self.http()?, &request).await
        })
    }

    fn place_order(&self, request: &OrderRequest) -> BoxFuture<'_, Result<Order>> {
        let request = request.clone();
        Box::pin(
            async move { private::place_order(self.credentials()?, self.http()?, &request).await },
        )
    }

    fn cancel_order(&self, market: &Market, order_id: &str) -> BoxFuture<'_, Result<()>> {
        let market = market.clone();
        let order_id = order_id.to_string();
        Box::pin(async move {
            private::cancel_order(self.credentials()?, self.http()?, &market, &order_id).await
        })
    }

    fn cancel_order_by_client_id(
        &self,
        market: &Market,
        client_id: &str,
    ) -> BoxFuture<'_, Result<()>> {
        let market = market.clone();
        let client_id = client_id.to_string();
        Box::pin(async move {
            private::cancel_order_by_client_id(
                self.credentials()?,
                self.http()?,
                &market,
                &client_id,
            )
            .await
        })
    }

    fn cancel_orders(
        &self,
        request: &CancelOrdersRequest,
    ) -> BoxFuture<'_, Result<CancelOrdersResult>> {
        let request = request.clone();
        Box::pin(async move {
            private::cancel_orders(self.credentials()?, self.http()?, &request).await
        })
    }

    fn subscribe_account(&self, config: &StreamConfig) -> BoxFuture<'_, Result<AccountStream>> {
        let url = format!("{}/private", self.region.websocket_url());
        let config = config.clone();

        Box::pin(async move {
            // The reconnect callback owns the credentials it signs with.
            let credentials = self.credentials()?.clone();
            let session = ws::connect(
                WsConnect {
                    url,
                    // Mint a fresh authorization value for every handshake.
                    headers: Some(Box::new(move || {
                        Ok(vec![(
                            private::AUTHORIZATION.to_string(),
                            private::authorization(&credentials, "")?,
                        )])
                    })),
                    subscribe: WsConnect::fixed(vec![private::subscribe_frame(&ticket())?]),
                    heartbeat: Some(stream::HEARTBEAT),
                },
                &config,
            )
            .await?;
            let close = session.close_handle();

            Ok(AccountStream::new_with_close(
                events(session, private::account_events, AccountEvent::Reconnected),
                move || async move { close.close().await },
            ))
        })
    }
}

/// Generates a unique subscription ticket.
fn ticket() -> String {
    uuid::Uuid::new_v4().to_string()
}

/// Decodes frames in arrival order and flattens each into zero or more events.
fn events<T: Clone + Send + 'static>(
    session: WsSession,
    mut decode: impl FnMut(&str) -> Result<Vec<T>> + Send + 'static,
    reconnected: T,
) -> impl Stream<Item = Result<T>> + Send {
    session.flat_map(move |item| {
        let items = match item {
            Ok(WsCommand::Text(text)) => split(decode(&text)),
            Ok(WsCommand::Binary(bytes)) => match String::from_utf8(bytes) {
                Ok(text) => split(decode(&text)),
                Err(err) => vec![Err(Error::decode(format!(
                    "upbit sent a frame that is not UTF-8: {err}"
                )))],
            },
            Ok(WsCommand::Reconnected) => vec![Ok(reconnected.clone())],
            Err(err) => vec![Err(err)],
        };

        futures_util::stream::iter(items)
    })
}

/// Converts one decoded frame into stream items.
fn split<T>(decoded: Result<Vec<T>>) -> Vec<Result<T>> {
    match decoded {
        Ok(items) => items.into_iter().map(Ok).collect(),
        Err(err) => vec![Err(err)],
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_spot_exchange_never_claims_derivatives_features() {
        let adapter = UpbitAdapter::new().with_credentials("access", "secret");

        for feature in [
            Feature::Positions,
            Feature::Margin,
            Feature::FundingRates,
            Feature::FundingPayments,
            Feature::MarginConfig,
            Feature::ReduceOnlyOrders,
        ] {
            assert!(!adapter.supports(feature), "{feature:?}");
        }
    }

    #[test]
    fn credentials_are_what_unlock_the_private_half() {
        let public = UpbitAdapter::new();
        let private = UpbitAdapter::new().with_credentials("access", "secret");

        for feature in [
            Feature::Balances,
            Feature::AssetNetworks,
            Feature::DepositAddresses,
            Feature::DepositHistory,
            Feature::DepositLookup,
            Feature::WithdrawalQuotes,
            Feature::Withdrawals,
            Feature::WithdrawalHistory,
            Feature::WithdrawalLookup,
            Feature::WithdrawalCancellation,
            Feature::Trading,
            Feature::AccountStream,
        ] {
            assert!(!public.supports(feature), "{feature:?}");
            assert!(private.supports(feature), "{feature:?}");
        }
    }

    #[test]
    fn public_market_data_works_without_credentials() {
        let public = UpbitAdapter::new();

        for feature in [
            Feature::Markets,
            Feature::Trades,
            Feature::OrderBook,
            Feature::Ticker,
            Feature::Candles,
            Feature::CandleStream,
        ] {
            assert!(public.supports(feature), "{feature:?}");
        }
    }

    #[test]
    fn travel_rule_requires_a_supported_region_and_credentials() {
        assert!(!UpbitAdapter::new().supports(Feature::TravelRule));
        assert!(!UpbitAdapter::with_region(UpbitRegion::Singapore).supports(Feature::TravelRule));
        assert!(
            UpbitAdapter::new()
                .with_credentials("access", "secret")
                .supports(Feature::TravelRule)
        );
        assert!(
            UpbitAdapter::with_region(UpbitRegion::Singapore)
                .with_credentials("access", "secret")
                .supports(Feature::TravelRule)
        );
        assert!(
            !UpbitAdapter::with_region(UpbitRegion::Indonesia)
                .with_credentials("access", "secret")
                .supports(Feature::TravelRule)
        );
    }

    #[tokio::test]
    async fn travel_rule_region_precedes_credential_validation() {
        let error = UpbitAdapter::with_region(UpbitRegion::Indonesia)
            .travel_rule_vasps()
            .await
            .expect_err("unsupported region must fail before credentials");
        assert!(matches!(
            error,
            Error::Unsupported {
                feature: Feature::TravelRule,
                ..
            }
        ));
    }

    #[tokio::test]
    async fn aggregated_order_books_fail_before_network_outside_korea() {
        let singapore = UpbitAdapter::with_region(UpbitRegion::Singapore);
        let market = Market::spot(Exchange::Upbit, "BTC", "SGD");

        assert!(matches!(
            singapore
                .order_books_at_level(&[market], Decimal::ONE, Some(1))
                .await,
            Err(Error::Unsupported {
                feature: Feature::OrderBook,
                ..
            })
        ));
    }

    #[tokio::test]
    async fn an_account_call_without_credentials_fails_before_the_network() {
        let public = UpbitAdapter::new();
        let market = Market::spot(Exchange::Upbit, "BTC", "KRW");
        let order = crate::request::OrderRequest::market(
            market.clone(),
            crate::types::Side::Sell,
            crate::types::Size::Base(rust_decimal::Decimal::ONE),
        );

        assert!(matches!(public.balances().await, Err(Error::Auth { .. })));
        assert!(matches!(
            public.open_orders(None).await,
            Err(Error::Auth { .. })
        ));
        assert!(matches!(
            public.place_order(&order).await,
            Err(Error::Auth { .. })
        ));
        assert!(matches!(
            public.test_order(&order).await,
            Err(Error::Auth { .. })
        ));
        assert!(matches!(
            public
                .cancel_and_new_order(&UpbitCancelAndNewOrderRequest::new(
                    UpbitOrderReference::uuid("order-1"),
                    UpbitCancelAndNewOrder::MarketSell {
                        volume: UpbitOrderVolume::RemainOnly,
                    },
                ))
                .await,
            Err(Error::Auth { .. })
        ));
        assert!(matches!(
            public
                .deposit_info("BTC", &crate::types::Network::Bitcoin)
                .await,
            Err(Error::Auth { .. })
        ));
        assert!(matches!(
            public
                .batch_cancel_open_orders(
                    &UpbitBatchCancelRequest::new(UpbitBatchCancelScope::All,)
                )
                .await,
            Err(Error::Auth { .. })
        ));
        assert!(matches!(
            public.cancel_order(&market, "an-order").await,
            Err(Error::Auth { .. })
        ));
        assert!(matches!(
            public.subscribe_account(&StreamConfig::default()).await,
            Err(Error::Auth { .. })
        ));
    }

    #[tokio::test]
    async fn the_derivatives_half_stays_at_the_trait_default() {
        let adapter = UpbitAdapter::new().with_credentials("access", "secret");
        let request =
            crate::request::HistoryRequest::new(Market::perpetual(Exchange::Upbit, "BTC", "KRW"));

        assert!(matches!(
            adapter.positions(None).await,
            Err(Error::Unsupported { .. })
        ));
        assert!(matches!(
            adapter.margin_summary().await,
            Err(Error::Unsupported { .. })
        ));
        assert!(matches!(
            adapter.funding_rates(&request).await,
            Err(Error::Unsupported { .. })
        ));
    }

    #[tokio::test]
    async fn upbit_lists_no_derivatives_and_says_so_with_an_empty_answer() {
        let markets = UpbitAdapter::new()
            .markets(MarketKind::Perpetual)
            .await
            .expect("a listable kind");

        assert!(markets.is_empty());
    }

    #[test]
    fn a_frame_that_carries_no_events_yields_none_and_a_bad_one_yields_one_error() {
        let mut decoder = stream::Decoder::default();

        assert!(
            decoder
                .decode(r#"{"status":"UP"}"#)
                .expect("a control frame")
                .is_empty()
        );
        assert_eq!(split(decoder.decode("not json")).len(), 1);
        assert_eq!(split(decoder.decode(r#"{"status":"UP"}"#)).len(), 0);
    }

    #[test]
    fn each_region_is_a_separate_deployment() {
        assert_eq!(UpbitAdapter::new().region(), UpbitRegion::Korea);
        assert_ne!(
            UpbitRegion::Korea.rest_base_url(),
            UpbitRegion::Singapore.rest_base_url()
        );
        assert!(UpbitRegion::Thailand.websocket_url().starts_with("wss://"));
    }

    #[tokio::test]
    async fn indonesia_external_withdrawal_fails_during_preparation_and_submission() {
        use crate::types::{ChainDestination, Network};
        use rust_decimal::Decimal;

        let request = WithdrawRequest::new(
            "BTC",
            Network::Bitcoin,
            Decimal::ONE,
            TransferDestination::Chain(ChainDestination {
                asset: "BTC".to_string(),
                network: Network::Bitcoin,
                address: "bc1destination".to_string(),
                memo: None,
            }),
        );

        let adapter = UpbitAdapter::with_region(UpbitRegion::Indonesia);
        for result in [
            adapter.prepare_withdrawal(&request).await.map(|_| ()),
            adapter.withdraw(&request).await.map(|_| ()),
        ] {
            assert!(matches!(
                result,
                Err(Error::Unsupported {
                    feature: Feature::Withdrawals,
                    ..
                })
            ));
        }
    }
}