nautilus_hyperliquid/websocket/

dispatch.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! WebSocket execution dispatch for the Hyperliquid execution client.
//!
//! Implements the two-tier execution dispatch contract from
//! `docs/developer_guide/adapters.md#tracked-and-external-execution-updates`:
//!
//! 1. The execution client registers an [`OrderIdentity`] in [`WsDispatchState`]
//!    when it submits an order, and refreshes the cached venue order id when a
//!    modify is sent so the WebSocket consumer can detect cancel-replace.
//! 2. Incoming [`OrderStatusReport`] and [`FillReport`] messages are routed
//!    through [`dispatch_order_event`] and [`dispatch_order_fill`].
//!    For tracked orders these build typed [`OrderEventAny`] events and emit
//!    them via [`ExecutionEventEmitter::send_order_event`]. For untracked /
//!    external orders the dispatch falls back to forwarding the raw report.
//!
//! The dispatch state lives in an `Arc<WsDispatchState>` shared between the
//! main client task (which registers identities at submission time) and the
//! spawned WebSocket consumer task.
//!
//! # GH-3827 cancel-replace handling
//!
//! Hyperliquid implements `modify` as a cancel-and-replace: the venue emits an
//! `ACCEPTED(new_voi)` together with a `CANCELED(old_voi)` under the same
//! `client_order_id`. The dispatch detects the replacement leg by comparing
//! `report.venue_order_id` to the last cached value, promotes it to an
//! `OrderUpdated` event, and suppresses the stale cancel so strategies never
//! observe a spurious termination.
//!
//! Each in-flight modify is tracked as an intent in a per-order chain (keyed
//! on `client_order_id`), pushed by `modify_order` before the HTTP call. An
//! intent lets dispatch skip an early `CANCELED(old_voi)` that arrives before
//! the replacement `ACCEPTED(new_voi)`, regardless of whether the WS message
//! races ahead of the HTTP response. Rapid repeated modifies under one stable
//! CLOID queue as a chain so a later modify cannot overwrite an earlier
//! intent's old-leg suppression, and a failed modify clears only its own
//! generation (leaving newer intents intact). The front intent is claimed on
//! promotion, advancing the next intent's old leg to the promoted replacement;
//! a rejected front reparents the next intent to the same still-live leg.
//!
//! A fill carrying the replacement `venue_order_id` during an in-flight modify
//! promotes the binding directly (the same `OrderUpdated` path as the
//! replacement `ACCEPTED`), so a dropped `ACCEPTED` does not strand the fill.
//! A fill is buffered into [`WsDispatchState::buffered_fills`] only when the
//! identity has no price to promote with; `handle_accepted` drains the buffer
//! on the replacement `ACCEPTED`. A delayed earlier-leg fill during a chained
//! modify is a known limitation. See GH-3972.
//!
//! When neither the replacement `ACCEPTED` nor a fill arrives, a query that
//! resolves the replacement by `cloid` promotes the binding the same way via
//! [`promote_replacement_from_query`], so a dropped `ACCEPTED` with no fill
//! cannot leave the order bound to the canceled leg.

use std::{
    collections::VecDeque,
    hash::Hash,
    sync::{
        Mutex,
        atomic::{AtomicBool, Ordering},
    },
};

use ahash::AHashSet;
use dashmap::{DashMap, DashSet};
use nautilus_core::{MUTEX_POISONED, UUID4, UnixNanos};
use nautilus_live::ExecutionEventEmitter;
use nautilus_model::{
    enums::{OrderSide, OrderStatus, OrderType},
    events::{
        OrderAccepted, OrderCanceled, OrderEventAny, OrderExpired, OrderFilled, OrderRejected,
        OrderTriggered, OrderUpdated,
    },
    identifiers::{AccountId, ClientOrderId, InstrumentId, StrategyId, TradeId, VenueOrderId},
    reports::{FillReport, OrderStatusReport},
    types::{Price, Quantity},
};
use ustr::Ustr;

use crate::{
    common::consts::HYPERLIQUID_POST_ONLY_WOULD_MATCH,
    http::models::HyperliquidExecPlaceOrderRequest,
};

pub const DEDUP_CAPACITY: usize = 10_000;

/// Identity metadata captured when an order is submitted through this client.
///
/// Stored in [`WsDispatchState::order_identities`] keyed by the full Nautilus
/// [`ClientOrderId`]. The dispatch functions use the identity to build typed
/// order events for tracked orders without needing access to the engine cache
/// (which is `!Send` and unreachable from the spawned WebSocket task).
#[derive(Debug, Clone)]
pub struct OrderIdentity {
    /// Strategy that owns the order.
    pub strategy_id: StrategyId,
    /// Instrument the order targets.
    pub instrument_id: InstrumentId,
    /// Order side captured at submission.
    pub order_side: OrderSide,
    /// Order type captured at submission.
    pub order_type: OrderType,
    /// Order quantity captured at submission.
    pub quantity: Quantity,
    /// Last known order price. Populated on submission and refreshed from
    /// subsequent status reports so a cancel-replace `ACCEPTED` that omits
    /// `price` can still produce an `OrderUpdated` carrying an accurate value.
    pub price: Option<Price>,
}

/// Bounded FIFO deduplication set.
///
/// When the capacity is reached, the oldest entry is evicted on the next
/// insert. A simple `clear()` at the threshold would drop every recent trade
/// id at once, opening a window where a reconnect or replay right after the
/// rollover could re-emit duplicate `OrderFilled` events; the FIFO window
/// slides instead.
#[derive(Debug)]
pub struct BoundedDedup<T>
where
    T: Eq + Hash + Clone,
{
    order: VecDeque<T>,
    set: AHashSet<T>,
    capacity: usize,
}

impl<T> BoundedDedup<T>
where
    T: Eq + Hash + Clone,
{
    /// Creates a new bounded dedup set with the given `capacity`.
    #[must_use]
    pub fn new(capacity: usize) -> Self {
        Self {
            order: VecDeque::with_capacity(capacity),
            set: AHashSet::with_capacity(capacity),
            capacity,
        }
    }

    /// Inserts a value. Returns `true` when the value was already present.
    pub fn insert(&mut self, value: T) -> bool {
        if self.set.contains(&value) {
            return true;
        }

        if self.order.len() >= self.capacity
            && let Some(evicted) = self.order.pop_front()
        {
            self.set.remove(&evicted);
        }

        self.order.push_back(value.clone());
        self.set.insert(value);
        false
    }

    /// Returns the number of entries currently tracked.
    #[must_use]
    pub fn len(&self) -> usize {
        self.set.len()
    }

    /// Returns whether the dedup set is empty.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.set.is_empty()
    }

    /// Returns whether the value is currently tracked.
    #[must_use]
    pub fn contains(&self, value: &T) -> bool {
        self.set.contains(value)
    }
}

/// Maximum in-flight modify intents tracked per order. Rapid repricing rarely
/// queues more than one or two unacknowledged modifies at once; the cap bounds
/// memory if replacement acks stall. On overflow the oldest intent is evicted.
pub const MAX_PENDING_MODIFY_INTENTS: usize = 32;

/// A single in-flight Hyperliquid modify awaiting its replacement leg.
///
/// Rapid repeated modifies under one stable CLOID queue as a chain of intents
/// so a later modify cannot overwrite an earlier pending old-leg marker, and a
/// failed modify clears only its own generation rather than a newer one's
/// state. Each intent carries the venue leg it cancel-replaces
/// (`old_venue_order_id`), the user-intended absolute total quantity, and the
/// exact request sent (used to size a corrective reduce).
#[derive(Debug, Clone)]
pub struct ModifyIntent {
    /// Monotonic per-order generation, used to clear a specific intent on failure.
    pub generation: u64,
    /// Venue order id this modify cancel-replaces, once known.
    pub old_venue_order_id: Option<VenueOrderId>,
    /// User-intended absolute total quantity for the replacement.
    pub target_qty: Quantity,
    /// Exact venue request sent, used to size a corrective reduce.
    pub sent_request: Option<HyperliquidExecPlaceOrderRequest>,
}

/// Bounded FIFO chain of in-flight modify intents for one order.
///
/// The venue processes chained modifies in submission order, so the front
/// (oldest) intent is the next to promote; on promotion the next intent's old
/// leg advances to the replacement just accepted.
#[derive(Debug, Default)]
struct ModifyChain {
    intents: VecDeque<ModifyIntent>,
    next_generation: u64,
}

/// Per-client dispatch state shared between order submission and the
/// WebSocket consumer task.
///
/// Tracks which orders were submitted through this client (so we can route
/// venue events to typed [`OrderEventAny`] emissions for tracked orders and
/// fall back to reports for external orders), provides cross-stream dedup
/// for `OrderAccepted` and `OrderFilled` emissions, and carries the
/// GH-3827 cancel-replace state (`cached_venue_order_ids` and
/// `pending_modify_keys`).
#[derive(Debug)]
pub struct WsDispatchState {
    /// Tracked orders keyed by full Nautilus [`ClientOrderId`].
    pub order_identities: DashMap<ClientOrderId, OrderIdentity>,
    /// Client order IDs for which an `OrderAccepted` event has been emitted.
    pub emitted_accepted: DashSet<ClientOrderId>,
    /// Tracked submissions whose POST response has not resolved yet.
    pending_submissions: DashSet<ClientOrderId>,
    /// Submission-time rejections held until the POST path can preserve its
    /// more detailed venue error string.
    pending_submission_rejections: DashMap<ClientOrderId, OrderStatusReport>,
    /// Client order IDs that have reached the filled terminal state.
    ///
    /// Retained past `cleanup_terminal` so that late replay of the same
    /// status or fill does not re-emit events.
    pub filled_orders: DashSet<ClientOrderId>,
    /// Trade IDs for which an `OrderFilled` event has been emitted.
    ///
    /// Bounded FIFO dedup to bound memory while keeping recent trade ids
    /// deduped across reconnects.
    pub emitted_trades: Mutex<BoundedDedup<TradeId>>,
    /// Raw Hyperliquid CLOIDs that reached a terminal state through the post
    /// response path before the matching `orderUpdates` event arrived.
    pub terminal_cloids: Mutex<BoundedDedup<Ustr>>,
    /// Last venue order id observed for a tracked client order id.
    ///
    /// Populated on the first `OrderAccepted` and refreshed on every
    /// cancel-replace promotion. A later `ACCEPTED` with a different venue
    /// order id under the same client order id is treated as the
    /// replacement leg of a Hyperliquid modify and emitted as `OrderUpdated`.
    pub cached_venue_order_ids: DashMap<ClientOrderId, VenueOrderId>,
    /// Per-order chain of in-flight modify intents, keyed by `client_order_id`.
    ///
    /// Rapid repeated modifies under one stable CLOID queue as a chain so a
    /// later modify cannot overwrite an earlier pending old-leg marker, and a
    /// failed modify clears only its own generation rather than a newer one's
    /// state. Populated by `modify_order` before the HTTP call so the WS cancel
    /// handler sees an intent even when `CANCELED(old_voi)` arrives before the
    /// HTTP response. A `CANCELED(old_voi)` matching any queued intent's old
    /// leg is suppressed so the later `ACCEPTED(new_voi)` can flow through the
    /// `OrderUpdated` path; the front intent is claimed on promotion and the
    /// next intent's old leg advances to the promoted replacement.
    pending_modify_chains: DashMap<ClientOrderId, ModifyChain>,
    /// `FillReport`s buffered only when a cancel-replace fill cannot be promoted
    /// (the identity carries no price); drained by the cancel-replace branch of
    /// `handle_accepted`. The common path promotes on the fill instead. See
    /// GH-3972.
    pub buffered_fills: DashMap<ClientOrderId, Vec<FillReport>>,
    /// Cumulative filled quantity per tracked order. Compared against
    /// `OrderIdentity::quantity` to decide when to clean up tracked state.
    pub order_filled_qty: DashMap<ClientOrderId, Quantity>,
    /// Corrective reduce queued by the cancel-replace promotion: client order
    /// id to (new venue order id, reduced request). Drained by the WS loop.
    pub pending_corrective: DashMap<ClientOrderId, (u64, HyperliquidExecPlaceOrderRequest)>,
    clearing: AtomicBool,
}

impl Default for WsDispatchState {
    fn default() -> Self {
        Self {
            order_identities: DashMap::new(),
            emitted_accepted: DashSet::default(),
            pending_submissions: DashSet::default(),
            pending_submission_rejections: DashMap::new(),
            filled_orders: DashSet::default(),
            emitted_trades: Mutex::new(BoundedDedup::new(DEDUP_CAPACITY)),
            terminal_cloids: Mutex::new(BoundedDedup::new(DEDUP_CAPACITY)),
            cached_venue_order_ids: DashMap::new(),
            pending_modify_chains: DashMap::new(),
            buffered_fills: DashMap::new(),
            order_filled_qty: DashMap::new(),
            pending_corrective: DashMap::new(),
            clearing: AtomicBool::new(false),
        }
    }
}

impl WsDispatchState {
    /// Creates a new empty dispatch state.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Registers an order identity. Called by the execution client at order
    /// submission time, before any WebSocket events for the order can arrive.
    pub fn register_identity(&self, client_order_id: ClientOrderId, identity: OrderIdentity) {
        self.order_identities.insert(client_order_id, identity);
    }

    /// Returns a clone of the identity for the given client order id, if any.
    #[must_use]
    pub fn lookup_identity(&self, client_order_id: &ClientOrderId) -> Option<OrderIdentity> {
        self.order_identities
            .get(client_order_id)
            .map(|r| r.clone())
    }

    /// Marks a tracked order as awaiting its submission POST response.
    pub fn mark_submission_pending(&self, client_order_id: ClientOrderId) {
        self.pending_submissions.insert(client_order_id);
    }

    /// Returns whether the order still awaits its submission POST response.
    #[must_use]
    pub fn submission_pending(&self, client_order_id: &ClientOrderId) -> bool {
        self.pending_submissions.contains(client_order_id)
    }

    /// Holds a submission-time rejection until the POST response resolves.
    pub fn buffer_submission_rejection(
        &self,
        client_order_id: ClientOrderId,
        report: OrderStatusReport,
    ) {
        self.pending_submission_rejections
            .insert(client_order_id, report);
    }

    /// Resolves submission tracking and returns any early rejection report.
    #[must_use]
    pub fn resolve_submission(&self, client_order_id: &ClientOrderId) -> Option<OrderStatusReport> {
        self.pending_submissions.remove(client_order_id);
        self.pending_submission_rejections
            .remove(client_order_id)
            .map(|(_, report)| report)
    }

    /// Refreshes the tracked price for a modify ack when the new report
    /// carries an updated price.
    pub fn update_identity_price(&self, client_order_id: &ClientOrderId, price: Option<Price>) {
        if let Some(price) = price
            && let Some(mut entry) = self.order_identities.get_mut(client_order_id)
        {
            entry.price = Some(price);
        }
    }

    /// Refreshes the tracked quantity for a modify ack.
    pub fn update_identity_quantity(&self, client_order_id: &ClientOrderId, quantity: Quantity) {
        if let Some(mut entry) = self.order_identities.get_mut(client_order_id) {
            entry.quantity = quantity;
        }
    }

    /// Marks an `OrderAccepted` event as emitted for this order.
    pub fn insert_accepted(&self, cid: ClientOrderId) {
        self.evict_if_full(&self.emitted_accepted);
        self.emitted_accepted.insert(cid);
    }

    /// Marks an order as having reached a terminal state.
    ///
    /// Returns `true` when this call claimed the terminal state, and `false`
    /// when another path had already claimed it.
    pub fn insert_filled(&self, cid: ClientOrderId) -> bool {
        self.evict_if_full(&self.filled_orders);
        self.filled_orders.insert(cid)
    }

    /// Atomically inserts a trade id into the dedup set.
    ///
    /// Returns `true` when the trade was already present (i.e. it is a
    /// duplicate), `false` otherwise.
    #[allow(
        clippy::missing_panics_doc,
        reason = "dedup mutex poisoning is not expected"
    )]
    pub fn check_and_insert_trade(&self, trade_id: TradeId) -> bool {
        let mut set = self.emitted_trades.lock().expect(MUTEX_POISONED);
        set.insert(trade_id)
    }

    /// Records a terminal raw Hyperliquid CLOID.
    ///
    /// Used when the post response rejects an order before the WebSocket
    /// `orderUpdates` message. The normal CLOID mapping can be removed while a
    /// late unresolved order update still gets suppressed instead of forwarded
    /// as an external report.
    #[allow(
        clippy::missing_panics_doc,
        reason = "terminal cloid mutex poisoning is not expected"
    )]
    pub fn insert_terminal_cloid(&self, cloid: Ustr) {
        let mut set = self.terminal_cloids.lock().expect(MUTEX_POISONED);
        set.insert(cloid);
    }

    /// Returns whether a raw Hyperliquid CLOID reached a terminal state through
    /// the post response path.
    #[allow(
        clippy::missing_panics_doc,
        reason = "terminal cloid mutex poisoning is not expected"
    )]
    #[must_use]
    pub fn terminal_cloid_seen(&self, cloid: &Ustr) -> bool {
        let set = self.terminal_cloids.lock().expect(MUTEX_POISONED);
        set.contains(cloid)
    }

    /// Caches the venue order id observed for a tracked client order id.
    pub fn record_venue_order_id(
        &self,
        client_order_id: ClientOrderId,
        venue_order_id: VenueOrderId,
    ) {
        self.cached_venue_order_ids
            .insert(client_order_id, venue_order_id);
    }

    /// Returns the previously cached venue order id, if any.
    #[must_use]
    pub fn cached_venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<VenueOrderId> {
        self.cached_venue_order_ids.get(client_order_id).map(|r| *r)
    }

    /// Queues an in-flight modify intent for cancel-before-accept suppression
    /// and records the target absolute total qty for the cancel-replace
    /// promotion. Returns the intent's generation.
    ///
    /// The generation lets the submission path clear only this modify on
    /// failure via [`Self::clear_modify_generation`], leaving newer queued
    /// modifies intact. Chained modifies append rather than overwrite, so a
    /// later modify cannot drop an earlier pending old-leg marker.
    pub fn mark_pending_modify(
        &self,
        client_order_id: ClientOrderId,
        old_venue_order_id: VenueOrderId,
        target_qty: Quantity,
    ) -> u64 {
        let mut chain = self
            .pending_modify_chains
            .entry(client_order_id)
            .or_default();
        let generation = chain.next_generation;
        chain.next_generation += 1;
        chain.intents.push_back(ModifyIntent {
            generation,
            old_venue_order_id: Some(old_venue_order_id),
            target_qty,
            sent_request: None,
        });

        if chain.intents.len() > MAX_PENDING_MODIFY_INTENTS {
            chain.intents.pop_front();
            log::warn!(
                "Modify chain for {client_order_id} exceeded {MAX_PENDING_MODIFY_INTENTS}; \
                 evicting oldest intent",
            );
        }
        generation
    }

    /// Clears the entire pending modify chain for a client order id.
    pub fn clear_pending_modify(&self, client_order_id: &ClientOrderId) {
        self.pending_modify_chains.remove(client_order_id);
    }

    /// Removes a single modify intent by generation, leaving newer queued
    /// modifies intact. Drops the chain entry when it empties.
    ///
    /// When the removed intent is the front, the next queued modify inherits
    /// its old leg: a rejected modify does not cancel-replace, so the resting
    /// leg it targeted is still live and the next modify cancel-replaces the
    /// same one. A non-front removal needs no reparenting; the front's
    /// promotion (or its own removal) advances the chain.
    pub fn clear_modify_generation(&self, client_order_id: &ClientOrderId, generation: u64) {
        let Some(mut chain) = self.pending_modify_chains.get_mut(client_order_id) else {
            return;
        };
        let removed_front_old = chain
            .intents
            .front()
            .filter(|front| front.generation == generation)
            .and_then(|front| front.old_venue_order_id);
        chain
            .intents
            .retain(|intent| intent.generation != generation);

        if let Some(old) = removed_front_old
            && let Some(new_front) = chain.intents.front_mut()
        {
            new_front.old_venue_order_id = Some(old);
        }
        drop(chain);
        // Remove only if still empty: a concurrent mark for the same order may
        // queue a new intent between the drop above and this remove
        self.pending_modify_chains
            .remove_if(client_order_id, |_, chain| chain.intents.is_empty());
    }

    /// Stashes the exact venue request sent onto the most recently queued
    /// modify intent for the order.
    pub fn stash_modify_request(
        &self,
        client_order_id: ClientOrderId,
        request: HyperliquidExecPlaceOrderRequest,
    ) {
        if let Some(mut chain) = self.pending_modify_chains.get_mut(&client_order_id)
            && let Some(back) = chain.intents.back_mut()
        {
            back.sent_request = Some(request);
        } else {
            log::debug!(
                "Stash modify request for {client_order_id} with no pending intent; ignoring"
            );
        }
    }

    /// Returns a clone of the front intent's stashed modify request, if any.
    #[must_use]
    pub fn modify_request(
        &self,
        client_order_id: &ClientOrderId,
    ) -> Option<HyperliquidExecPlaceOrderRequest> {
        self.pending_modify_chains
            .get(client_order_id)
            .and_then(|chain| chain.intents.front().and_then(|i| i.sent_request.clone()))
    }

    /// Claims the front (oldest) modify intent for promotion.
    ///
    /// Advances the next queued intent's old leg to `new_venue_order_id`: its
    /// cancel-replace targets the replacement just promoted, not the leg it was
    /// queued against. Returns the claimed intent, or `None` when no intent is
    /// queued (an external modify with no local marker). Drops the chain entry
    /// when it empties.
    pub fn claim_front_modify(
        &self,
        client_order_id: &ClientOrderId,
        new_venue_order_id: VenueOrderId,
    ) -> Option<ModifyIntent> {
        let mut chain = self.pending_modify_chains.get_mut(client_order_id)?;
        let claimed = chain.intents.pop_front();
        if let Some(next) = chain.intents.front_mut() {
            next.old_venue_order_id = Some(new_venue_order_id);
        }
        drop(chain);
        // Remove only if still empty: a concurrent mark for the same order may
        // queue a new intent between the drop above and this remove
        self.pending_modify_chains
            .remove_if(client_order_id, |_, chain| chain.intents.is_empty());
        claimed
    }

    /// Queues a corrective reduce for the WebSocket consumer loop to post.
    pub fn queue_corrective(
        &self,
        client_order_id: ClientOrderId,
        oid: u64,
        request: HyperliquidExecPlaceOrderRequest,
    ) {
        self.pending_corrective
            .insert(client_order_id, (oid, request));
    }

    /// Removes and returns a queued corrective reduce, if any.
    #[must_use]
    pub fn take_corrective(
        &self,
        client_order_id: &ClientOrderId,
    ) -> Option<(u64, HyperliquidExecPlaceOrderRequest)> {
        self.pending_corrective
            .remove(client_order_id)
            .map(|(_, v)| v)
    }

    /// Returns whether any modify intent is queued for the client order id.
    #[must_use]
    pub fn has_pending_modify(&self, client_order_id: &ClientOrderId) -> bool {
        self.pending_modify_chains
            .get(client_order_id)
            .is_some_and(|chain| !chain.intents.is_empty())
    }

    /// Returns the front intent's old venue order id, if any.
    #[must_use]
    pub fn pending_modify(&self, client_order_id: &ClientOrderId) -> Option<VenueOrderId> {
        self.pending_modify_chains
            .get(client_order_id)
            .and_then(|chain| chain.intents.front().and_then(|i| i.old_venue_order_id))
    }

    /// Returns whether any queued intent cancel-replaces `venue_order_id`.
    ///
    /// Used to suppress the `CANCELED(old_voi)` leg of any in-flight modify in
    /// the chain, not only the oldest.
    #[must_use]
    pub fn pending_modify_contains_old(
        &self,
        client_order_id: &ClientOrderId,
        venue_order_id: VenueOrderId,
    ) -> bool {
        self.pending_modify_chains
            .get(client_order_id)
            .is_some_and(|chain| {
                chain
                    .intents
                    .iter()
                    .any(|i| i.old_venue_order_id == Some(venue_order_id))
            })
    }

    /// Returns the front intent's recorded target absolute total qty, if any.
    #[must_use]
    pub fn pending_modify_target_qty(&self, client_order_id: &ClientOrderId) -> Option<Quantity> {
        self.pending_modify_chains
            .get(client_order_id)
            .and_then(|chain| chain.intents.front().map(|i| i.target_qty))
    }

    /// Buffers a `FillReport` arrived during an in-flight cancel-replace.
    pub fn buffer_fill(&self, client_order_id: ClientOrderId, fill: FillReport) {
        self.buffered_fills
            .entry(client_order_id)
            .or_default()
            .push(fill);
    }

    /// Removes and returns buffered fills for the cid, in arrival order.
    #[must_use]
    pub fn drain_buffered_fills(&self, client_order_id: &ClientOrderId) -> Vec<FillReport> {
        self.buffered_fills
            .remove(client_order_id)
            .map(|(_, v)| v)
            .unwrap_or_default()
    }

    /// Number of buffered fills for the cid.
    #[must_use]
    pub fn buffered_fill_count(&self, client_order_id: &ClientOrderId) -> usize {
        self.buffered_fills
            .get(client_order_id)
            .map_or(0, |r| r.len())
    }

    /// Records cumulative filled quantity for a tracked order.
    pub fn record_filled_qty(&self, client_order_id: ClientOrderId, qty: Quantity) {
        self.order_filled_qty.insert(client_order_id, qty);
    }

    /// Returns the previously recorded cumulative filled quantity, if any.
    #[must_use]
    pub fn previous_filled_qty(&self, client_order_id: &ClientOrderId) -> Option<Quantity> {
        self.order_filled_qty.get(client_order_id).map(|r| *r)
    }

    /// Removes all dispatch state for an order that has reached a terminal state.
    ///
    /// `filled_orders` is intentionally *not* cleared here: the marker is
    /// used to suppress stale replays and must outlive the identity cleanup.
    pub fn cleanup_terminal(&self, client_order_id: &ClientOrderId) {
        self.order_identities.remove(client_order_id);
        self.emitted_accepted.remove(client_order_id);
        self.pending_submissions.remove(client_order_id);
        self.pending_submission_rejections.remove(client_order_id);
        self.cached_venue_order_ids.remove(client_order_id);
        self.pending_modify_chains.remove(client_order_id);
        self.pending_corrective.remove(client_order_id);
        self.buffered_fills.remove(client_order_id);
        self.order_filled_qty.remove(client_order_id);
    }

    fn evict_if_full(&self, set: &DashSet<ClientOrderId>) {
        if set.len() >= DEDUP_CAPACITY
            && self
                .clearing
                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
                .is_ok()
        {
            set.clear();
            self.clearing.store(false, Ordering::Release);
        }
    }
}

/// Outcome of a single dispatch call.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DispatchOutcome {
    /// The report was for a tracked order. Typed events have been emitted
    /// (or intentionally skipped, e.g. dedup hit). The caller must not
    /// forward the report as a fallback.
    Tracked,
    /// The report is for an external / untracked order. The caller should
    /// forward the report via [`ExecutionEventEmitter::send_order_status_report`]
    /// or [`ExecutionEventEmitter::send_fill_report`] so the engine can
    /// reconcile.
    External,
    /// The report was recognised as stale (e.g. cancel leg of a
    /// cancel-replace modify, or replay after terminal state). The caller
    /// must drop it without forwarding.
    Skip,
}

/// Dispatches an [`OrderStatusReport`] using the two-tier routing contract.
///
/// Returns [`DispatchOutcome::Tracked`] when the report maps to a tracked
/// order (typed events have been emitted or dedup hit), [`External`] when
/// the caller should forward the report as an untracked fallback, or
/// [`Skip`] when the report is a stale / race leg that must be dropped.
///
/// [`External`]: DispatchOutcome::External
/// [`Skip`]: DispatchOutcome::Skip
pub fn dispatch_order_event(
    report: &OrderStatusReport,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    let Some(client_order_id) = report.client_order_id else {
        return DispatchOutcome::External;
    };

    if state.filled_orders.contains(&client_order_id) {
        log::debug!(
            "Skipping stale report for filled order: cid={client_order_id}, status={:?}",
            report.order_status,
        );
        return DispatchOutcome::Skip;
    }

    let client_order_id_str = client_order_id.as_str();
    if client_order_id_str.starts_with("0x")
        && state.terminal_cloid_seen(&Ustr::from(client_order_id_str))
    {
        log::debug!(
            "Skipping stale terminal report for raw cloid: cid={client_order_id}, status={:?}",
            report.order_status,
        );
        return DispatchOutcome::Skip;
    }

    let Some(identity) = state.lookup_identity(&client_order_id) else {
        return DispatchOutcome::External;
    };

    match report.order_status {
        OrderStatus::Accepted => {
            handle_accepted(report, client_order_id, &identity, state, emitter, ts_init)
        }
        OrderStatus::Triggered => {
            handle_triggered(report, client_order_id, &identity, state, emitter, ts_init)
        }
        OrderStatus::Canceled => {
            handle_canceled(report, client_order_id, &identity, state, emitter, ts_init)
        }
        OrderStatus::Expired => {
            handle_expired(report, client_order_id, &identity, state, emitter, ts_init)
        }
        OrderStatus::Rejected => {
            handle_rejected(report, client_order_id, &identity, state, emitter, ts_init)
        }
        OrderStatus::Filled => handle_filled_marker(client_order_id, state),
        OrderStatus::PartiallyFilled => {
            // Fills come via `FillReport`; nothing to emit from the status path.
            DispatchOutcome::Tracked
        }
        OrderStatus::PendingUpdate
        | OrderStatus::PendingCancel
        | OrderStatus::Submitted
        | OrderStatus::Initialized
        | OrderStatus::Denied
        | OrderStatus::Released
        | OrderStatus::Emulated
        | OrderStatus::Voided => DispatchOutcome::Tracked,
    }
}

/// Dispatches a [`FillReport`] using the two-tier routing contract.
///
/// Returns [`DispatchOutcome::Tracked`] when the fill has been emitted as
/// an `OrderFilled` event (or skipped via trade dedup), [`External`] when
/// the caller should forward the fill via
/// [`ExecutionEventEmitter::send_fill_report`], or [`Skip`] when the fill
/// is a replay for an already-terminal order and must be dropped.
///
/// [`External`]: DispatchOutcome::External
/// [`Skip`]: DispatchOutcome::Skip
pub fn dispatch_order_fill(
    report: &FillReport,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    let Some(client_order_id) = report.client_order_id else {
        return DispatchOutcome::External;
    };

    if state.filled_orders.contains(&client_order_id) {
        log::debug!(
            "Skipping stale fill for filled order: cid={client_order_id}, trade_id={}",
            report.trade_id,
        );
        return DispatchOutcome::Skip;
    }

    let Some(mut identity) = state.lookup_identity(&client_order_id) else {
        return DispatchOutcome::External;
    };

    // Set when a fill promotes, so the corrective-reduce runs after the fill applies
    let mut promoted_corrective: Option<(Quantity, HyperliquidExecPlaceOrderRequest)> = None;

    // Promote the binding from the fill so a dropped replacement ACCEPTED cannot
    // strand it (see module docs).
    if state.has_pending_modify(&client_order_id)
        && let Some(cached_voi) = state.cached_venue_order_id(&client_order_id)
        && report.venue_order_id != cached_voi
    {
        let target = state.pending_modify_target_qty(&client_order_id);
        let sent_request = state.modify_request(&client_order_id);
        // Prefer the modify target price over the stale cached identity price
        let price = sent_request
            .as_ref()
            .zip(identity.price)
            .and_then(|(r, cached)| Price::from_decimal_dp(r.price, cached.precision).ok())
            .or(identity.price);
        let Some(price) = price else {
            log::warn!(
                "Cannot promote cancel-replace for {client_order_id} from fill: no target \
                 or cached price; buffering until the replacement ACCEPTED arrives",
            );
            state.buffer_fill(client_order_id, report.clone());
            return DispatchOutcome::Tracked;
        };
        let updated_quantity = target.unwrap_or(identity.quantity);
        promote_cancel_replace(
            client_order_id,
            &identity,
            state,
            emitter,
            report.venue_order_id,
            report.account_id,
            price,
            updated_quantity,
            None,
            report.ts_event,
            ts_init,
        );
        // Re-read the identity advanced by the promotion (quantity and price)
        if let Some(updated) = state.lookup_identity(&client_order_id) {
            identity = updated;
        }

        if let (Some(target), Some(sent_request)) = (target, sent_request) {
            promoted_corrective = Some((target, sent_request));
        }
    }

    if state.check_and_insert_trade(report.trade_id) {
        log::debug!(
            "Skipping duplicate fill for {client_order_id}: trade_id={}",
            report.trade_id
        );
        return DispatchOutcome::Tracked;
    }

    let previous = state
        .previous_filled_qty(&client_order_id)
        .unwrap_or_else(|| Quantity::zero(report.last_qty.precision));
    let cumulative = previous + report.last_qty;

    let is_terminal_fill = cumulative >= identity.quantity;
    if is_terminal_fill && !claim_terminal_order(client_order_id, state, OrderStatus::Filled) {
        return DispatchOutcome::Skip;
    }

    ensure_accepted_emitted(
        client_order_id,
        report.venue_order_id,
        report.account_id,
        &identity,
        state,
        emitter,
        report.ts_event,
        ts_init,
    );

    let filled = OrderFilled::new(
        emitter.trader_id(),
        identity.strategy_id,
        identity.instrument_id,
        client_order_id,
        report.venue_order_id,
        report.account_id,
        report.trade_id,
        identity.order_side,
        identity.order_type,
        report.last_qty,
        report.last_px,
        report.commission.currency,
        report.liquidity_side,
        UUID4::new(),
        report.ts_event,
        ts_init,
        false,
        report.venue_position_id,
        Some(report.commission),
        None,
    );
    emitter.send_order_event(OrderEventAny::Filled(filled));

    state.record_filled_qty(client_order_id, cumulative);

    // Cumulative now includes this fill, so the reduce sizes against the true remaining
    if let Some((target, sent_request)) = promoted_corrective {
        maybe_queue_corrective_reduce(
            state,
            client_order_id,
            report.venue_order_id,
            target,
            sent_request,
        );
    }

    if is_terminal_fill {
        state.cleanup_terminal(&client_order_id);
    }

    DispatchOutcome::Tracked
}

fn handle_accepted(
    report: &OrderStatusReport,
    client_order_id: ClientOrderId,
    identity: &OrderIdentity,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    let venue_order_id = report.venue_order_id;
    let ts_event = report.ts_last;
    let account_id = report.account_id;

    // Cancel-replace detection: if an earlier ACCEPTED cached a different
    // venue_order_id under the same client_order_id, this ACCEPTED is the
    // replacement leg of a Hyperliquid modify and must be promoted to
    // OrderUpdated. See GH-3827.
    if let Some(cached_voi) = state.cached_venue_order_id(&client_order_id)
        && cached_voi != venue_order_id
    {
        let price = report.price.or(identity.price);
        let Some(price) = price else {
            log::warn!(
                "Cannot emit OrderUpdated for cancel-replace {client_order_id}: \
                 no price on report and no cached price on identity",
            );
            return DispatchOutcome::Skip;
        };

        // Prefer user target over venue's remaining-only `report.quantity`;
        // fall back when no marker (external modify).
        let target_total_qty = state.pending_modify_target_qty(&client_order_id);
        let updated_quantity = target_total_qty.unwrap_or(report.quantity);
        let sent_request = state.modify_request(&client_order_id);

        promote_cancel_replace(
            client_order_id,
            identity,
            state,
            emitter,
            venue_order_id,
            account_id,
            price,
            updated_quantity,
            report.trigger_price,
            ts_event,
            ts_init,
        );

        if let (Some(target), Some(sent_request)) = (target_total_qty, sent_request) {
            maybe_queue_corrective_reduce(
                state,
                client_order_id,
                venue_order_id,
                target,
                sent_request,
            );
        }

        return DispatchOutcome::Tracked;
    }

    if state.emitted_accepted.contains(&client_order_id) {
        // Repeat ACCEPTED for an already-accepted order. Nothing to emit;
        // refresh the cached price so a subsequent cancel-replace without a
        // report price can still recover an accurate value.
        state.update_identity_price(&client_order_id, report.price);
        return DispatchOutcome::Tracked;
    }

    state.insert_accepted(client_order_id);
    state.record_venue_order_id(client_order_id, venue_order_id);
    state.update_identity_price(&client_order_id, report.price);

    let accepted = OrderAccepted::new(
        emitter.trader_id(),
        identity.strategy_id,
        identity.instrument_id,
        client_order_id,
        venue_order_id,
        account_id,
        UUID4::new(),
        ts_event,
        ts_init,
        false,
    );
    emitter.send_order_event(OrderEventAny::Accepted(accepted));
    DispatchOutcome::Tracked
}

// Shared by the ACCEPTED branch and the fill path (dropped-ACCEPTED recovery) so the
// cancel-replace binding is recovered from whichever arrives first. See GH-3827, GH-3972.
#[allow(
    clippy::too_many_arguments,
    reason = "promotion needs the full OrderUpdated field set, sourced from two report shapes"
)]
fn promote_cancel_replace(
    client_order_id: ClientOrderId,
    identity: &OrderIdentity,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    venue_order_id: VenueOrderId,
    account_id: AccountId,
    price: Price,
    quantity: Quantity,
    trigger_price: Option<Price>,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) {
    state.record_venue_order_id(client_order_id, venue_order_id);
    state.update_identity_quantity(&client_order_id, quantity);
    state.update_identity_price(&client_order_id, Some(price));
    // Claim the front intent; the next queued modify advances to this replacement
    state.claim_front_modify(&client_order_id, venue_order_id);

    let updated = OrderUpdated::new(
        emitter.trader_id(),
        identity.strategy_id,
        identity.instrument_id,
        client_order_id,
        quantity,
        UUID4::new(),
        ts_event,
        ts_init,
        false,
        Some(venue_order_id),
        Some(account_id),
        Some(price),
        trigger_price,
        None,
        false,
    );
    emitter.send_order_event(OrderEventAny::Updated(updated));

    // Drain fills buffered before the binding advanced. Bypasses
    // `handle_execution_report`; FIFO-bounded caches make any residue benign.
    let buffered = state.drain_buffered_fills(&client_order_id);
    for fill in buffered {
        dispatch_order_fill(&fill, state, emitter, ts_init);
    }
}

/// Promotes a cancel-replace replacement surfaced by a query during an in-flight modify.
///
/// When the query returns the replacement leg (`Accepted`, `venue_order_id` diverging from the
/// cached one, modify tracked), emits the `OrderUpdated` that rebinds the order, so a dropped
/// replacement `Accepted` with no fill cannot strand the binding on the canceled leg. Returns
/// `true` when promoted; the caller still forwards the report so the engine confirms the order.
pub fn promote_replacement_from_query(
    report: &OrderStatusReport,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> bool {
    if report.order_status != OrderStatus::Accepted {
        return false;
    }

    let Some(client_order_id) = report.client_order_id else {
        return false;
    };

    if !state.has_pending_modify(&client_order_id) {
        return false;
    }

    let Some(cached_voi) = state.cached_venue_order_id(&client_order_id) else {
        return false;
    };

    if report.venue_order_id == cached_voi {
        return false;
    }

    let Some(identity) = state.lookup_identity(&client_order_id) else {
        return false;
    };

    let Some(price) = report.price.or(identity.price) else {
        log::warn!(
            "Cannot promote cancel-replace from query for {client_order_id}: \
             no price on report and no cached price on identity",
        );
        return false;
    };

    // Prefer the user target over the venue's remaining-only `report.quantity`
    let updated_quantity = state
        .pending_modify_target_qty(&client_order_id)
        .unwrap_or(report.quantity);

    promote_cancel_replace(
        client_order_id,
        &identity,
        state,
        emitter,
        report.venue_order_id,
        report.account_id,
        price,
        updated_quantity,
        report.trigger_price,
        report.ts_last,
        ts_init,
    );

    log::debug!("Promoted cancel-replace replacement for {client_order_id} from query");

    true
}

// Queue a corrective reduce when a fill that raced the modify left the replacement
// oversized. Reached from both promotion paths; the engine overfill guard backstops.
fn maybe_queue_corrective_reduce(
    state: &WsDispatchState,
    client_order_id: ClientOrderId,
    venue_order_id: VenueOrderId,
    target: Quantity,
    sent_request: HyperliquidExecPlaceOrderRequest,
) {
    let Ok(new_oid) = venue_order_id.as_str().parse::<u64>() else {
        return;
    };

    let filled = state
        .previous_filled_qty(&client_order_id)
        .unwrap_or_else(|| Quantity::zero(target.precision));
    if filled >= target {
        return;
    }

    let remaining = (target - filled).as_decimal().normalize();

    let sent_size = sent_request.size;
    if sent_size > remaining {
        let mut corrective = sent_request;
        corrective.size = remaining;

        state.mark_pending_modify(client_order_id, venue_order_id, target);
        state.stash_modify_request(client_order_id, corrective.clone());
        state.queue_corrective(client_order_id, new_oid, corrective);

        log::warn!(
            "Cancel-replace left {client_order_id} oversized on {venue_order_id} \
             (sent {sent_size}, remaining {remaining}); queuing corrective reduce",
        );
    }
}

fn handle_triggered(
    report: &OrderStatusReport,
    client_order_id: ClientOrderId,
    identity: &OrderIdentity,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    if !matches!(
        identity.order_type,
        OrderType::StopLimit | OrderType::TrailingStopLimit | OrderType::LimitIfTouched
    ) {
        log::debug!(
            "Ignoring TRIGGERED status for non-triggerable order type {:?}: {client_order_id}",
            identity.order_type,
        );
        return DispatchOutcome::Tracked;
    }

    ensure_accepted_emitted(
        client_order_id,
        report.venue_order_id,
        report.account_id,
        identity,
        state,
        emitter,
        report.ts_last,
        ts_init,
    );

    let triggered = OrderTriggered::new(
        emitter.trader_id(),
        identity.strategy_id,
        identity.instrument_id,
        client_order_id,
        UUID4::new(),
        report.ts_last,
        ts_init,
        false,
        Some(report.venue_order_id),
        Some(report.account_id),
    );
    emitter.send_order_event(OrderEventAny::Triggered(triggered));
    DispatchOutcome::Tracked
}

fn handle_canceled(
    report: &OrderStatusReport,
    client_order_id: ClientOrderId,
    identity: &OrderIdentity,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    let venue_order_id = report.venue_order_id;

    // Stale cancel suppression: if the cached venue_order_id has already
    // been advanced by a cancel-replace promotion, this CANCELED refers to
    // the old leg and has already been handled as OrderUpdated. See GH-3827.
    if let Some(cached_voi) = state.cached_venue_order_id(&client_order_id)
        && cached_voi != venue_order_id
    {
        log::debug!(
            "Skipping stale CANCELED for {venue_order_id} (cached {cached_voi}) on {client_order_id}",
        );
        return DispatchOutcome::Skip;
    }

    // Cancel-before-accept race: an in-flight modify may deliver
    // CANCELED(old_voi) before the replacement ACCEPTED(new_voi). Any queued
    // intent whose old leg matches (marked before the HTTP call, cleared on
    // failure) suppresses that cancel so the later ACCEPTED routes through
    // OrderUpdated. See GH-3827.
    if state.pending_modify_contains_old(&client_order_id, venue_order_id) {
        log::debug!(
            "Skipping cancel-before-accept leg for {client_order_id}: venue_order_id={venue_order_id}",
        );
        return DispatchOutcome::Skip;
    }

    if !claim_terminal_order(client_order_id, state, report.order_status) {
        return DispatchOutcome::Skip;
    }

    ensure_accepted_emitted(
        client_order_id,
        venue_order_id,
        report.account_id,
        identity,
        state,
        emitter,
        report.ts_last,
        ts_init,
    );

    let canceled = OrderCanceled::new(
        emitter.trader_id(),
        identity.strategy_id,
        identity.instrument_id,
        client_order_id,
        UUID4::new(),
        report.ts_last,
        ts_init,
        false,
        Some(venue_order_id),
        Some(report.account_id),
    );
    emitter.send_order_event(OrderEventAny::Canceled(canceled));

    state.cleanup_terminal(&client_order_id);
    DispatchOutcome::Tracked
}

fn handle_expired(
    report: &OrderStatusReport,
    client_order_id: ClientOrderId,
    identity: &OrderIdentity,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    if !claim_terminal_order(client_order_id, state, report.order_status) {
        return DispatchOutcome::Skip;
    }

    ensure_accepted_emitted(
        client_order_id,
        report.venue_order_id,
        report.account_id,
        identity,
        state,
        emitter,
        report.ts_last,
        ts_init,
    );

    let expired = OrderExpired::new(
        emitter.trader_id(),
        identity.strategy_id,
        identity.instrument_id,
        client_order_id,
        UUID4::new(),
        report.ts_last,
        ts_init,
        false,
        Some(report.venue_order_id),
        Some(report.account_id),
    );
    emitter.send_order_event(OrderEventAny::Expired(expired));
    state.cleanup_terminal(&client_order_id);
    DispatchOutcome::Tracked
}

fn handle_rejected(
    report: &OrderStatusReport,
    client_order_id: ClientOrderId,
    identity: &OrderIdentity,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    if state.submission_pending(&client_order_id) {
        state.buffer_submission_rejection(client_order_id, report.clone());
        return DispatchOutcome::Skip;
    }

    if !claim_terminal_order(client_order_id, state, report.order_status) {
        return DispatchOutcome::Skip;
    }

    let reason = report
        .cancel_reason
        .clone()
        .unwrap_or_else(|| "Order rejected by exchange".to_string());
    let rejected = OrderRejected::new(
        emitter.trader_id(),
        identity.strategy_id,
        identity.instrument_id,
        client_order_id,
        report.account_id,
        Ustr::from(&reason),
        UUID4::new(),
        report.ts_last,
        ts_init,
        false,
        report.post_only && reason.contains(HYPERLIQUID_POST_ONLY_WOULD_MATCH),
    );
    emitter.send_order_event(OrderEventAny::Rejected(rejected));
    state.cleanup_terminal(&client_order_id);
    DispatchOutcome::Tracked
}

fn claim_terminal_order(
    client_order_id: ClientOrderId,
    state: &WsDispatchState,
    status: OrderStatus,
) -> bool {
    let claimed = state.insert_filled(client_order_id);
    if !claimed {
        log::debug!("Skipping duplicate terminal event for {client_order_id}: status={status:?}",);
    }

    claimed
}

fn handle_filled_marker(
    _client_order_id: ClientOrderId,
    _state: &WsDispatchState,
) -> DispatchOutcome {
    // A status-only `FILLED` marker does not carry fill data; the actual
    // `OrderFilled` is emitted from `dispatch_order_fill` when the matching
    // trade arrives. Do *not* set `filled_orders` here, otherwise the
    // follow-up fill would be classified as a stale replay and dropped
    // before the terminal `OrderFilled` event can be emitted. The fill
    // path installs the marker itself once the cumulative fill quantity
    // matches the tracked order quantity.
    DispatchOutcome::Tracked
}

/// Synthesizes and emits an `OrderAccepted` event when one has not yet been
/// emitted for the given order.
///
/// Used before emitting non-Accepted events so strategies always observe the
/// canonical `Submitted -> Accepted -> ...` lifecycle even when the venue
/// compresses the placement and follow-up event into a single message (fast
/// fills).
#[allow(clippy::too_many_arguments)]
fn ensure_accepted_emitted(
    client_order_id: ClientOrderId,
    venue_order_id: VenueOrderId,
    account_id: AccountId,
    identity: &OrderIdentity,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) {
    if state.emitted_accepted.contains(&client_order_id) {
        return;
    }
    state.insert_accepted(client_order_id);
    state.record_venue_order_id(client_order_id, venue_order_id);

    let accepted = OrderAccepted::new(
        emitter.trader_id(),
        identity.strategy_id,
        identity.instrument_id,
        client_order_id,
        venue_order_id,
        account_id,
        UUID4::new(),
        ts_event,
        ts_init,
        false,
    );
    emitter.send_order_event(OrderEventAny::Accepted(accepted));
}

#[cfg(test)]
mod tests {
    use nautilus_model::identifiers::{ClientOrderId, InstrumentId, StrategyId, TradeId};
    use rstest::rstest;
    use rust_decimal::Decimal;

    use super::*;
    use crate::http::models::{
        HyperliquidExecLimitParams, HyperliquidExecOrderKind, HyperliquidExecTif,
    };

    fn make_identity() -> OrderIdentity {
        OrderIdentity {
            strategy_id: StrategyId::from("S-001"),
            instrument_id: InstrumentId::from("BTC-USD-PERP.HYPERLIQUID"),
            order_side: OrderSide::Buy,
            order_type: OrderType::Limit,
            quantity: Quantity::from("0.0001"),
            price: None,
        }
    }

    #[rstest]
    fn test_register_and_lookup_identity() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-001");
        state.register_identity(cid, make_identity());

        let found = state.lookup_identity(&cid);
        assert!(found.is_some());
        let identity = found.unwrap();
        assert_eq!(identity.strategy_id.as_str(), "S-001");
        assert_eq!(identity.order_side, OrderSide::Buy);
    }

    #[rstest]
    fn test_lookup_identity_missing_returns_none() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("not-tracked");
        assert!(state.lookup_identity(&cid).is_none());
    }

    #[rstest]
    fn test_insert_accepted_dedup() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-002");
        assert!(!state.emitted_accepted.contains(&cid));
        state.insert_accepted(cid);
        assert!(state.emitted_accepted.contains(&cid));
        state.insert_accepted(cid);
        assert!(state.emitted_accepted.contains(&cid));
    }

    #[rstest]
    fn test_check_and_insert_trade_detects_duplicates() {
        let state = WsDispatchState::new();
        let trade = TradeId::new("trade-1");
        assert!(!state.check_and_insert_trade(trade));
        assert!(state.check_and_insert_trade(trade));
    }

    #[rstest]
    fn test_bounded_dedup_fifo_eviction_preserves_recent_ids() {
        let mut dedup: BoundedDedup<TradeId> = BoundedDedup::new(3);
        assert!(!dedup.insert(TradeId::new("t-0")));
        assert!(!dedup.insert(TradeId::new("t-1")));
        assert!(!dedup.insert(TradeId::new("t-2")));
        assert_eq!(dedup.len(), 3);

        // Overflow evicts the oldest.
        assert!(!dedup.insert(TradeId::new("t-3")));
        assert_eq!(dedup.len(), 3);
        assert!(!dedup.contains(&TradeId::new("t-0")));
        assert!(dedup.contains(&TradeId::new("t-1")));
        assert!(dedup.contains(&TradeId::new("t-3")));
    }

    #[rstest]
    fn test_pending_modify_roundtrip() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-010");
        let voi = VenueOrderId::new("v-1");
        let target_qty = Quantity::from("0.0001");

        assert!(state.pending_modify(&cid).is_none());
        assert!(state.pending_modify_target_qty(&cid).is_none());
        state.mark_pending_modify(cid, voi, target_qty);
        assert_eq!(state.pending_modify(&cid), Some(voi));
        assert_eq!(state.pending_modify_target_qty(&cid), Some(target_qty));
        state.clear_pending_modify(&cid);
        assert!(state.pending_modify(&cid).is_none());
        assert!(state.pending_modify_target_qty(&cid).is_none());
    }

    #[rstest]
    fn test_cleanup_terminal_preserves_filled_marker() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-020");
        state.register_identity(cid, make_identity());
        state.insert_accepted(cid);
        state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("0.0001"));
        state.insert_filled(cid);
        state.cleanup_terminal(&cid);

        assert!(state.lookup_identity(&cid).is_none());
        assert!(!state.emitted_accepted.contains(&cid));
        assert!(state.pending_modify(&cid).is_none());
        assert!(state.pending_modify_target_qty(&cid).is_none());
        // `filled_orders` outlives `cleanup_terminal` so replays stay suppressed.
        assert!(state.filled_orders.contains(&cid));
    }

    #[rstest]
    fn test_cleanup_terminal_clears_corrective_state() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-021");
        let request = sample_request(Decimal::from(1));
        state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("1"));
        state.stash_modify_request(cid, request.clone());
        state.queue_corrective(cid, 1, request);
        assert!(state.modify_request(&cid).is_some());

        state.cleanup_terminal(&cid);

        assert!(state.modify_request(&cid).is_none());
        assert!(state.take_corrective(&cid).is_none());
        assert!(state.pending_modify(&cid).is_none());
    }

    fn sample_request(size: Decimal) -> HyperliquidExecPlaceOrderRequest {
        HyperliquidExecPlaceOrderRequest {
            asset: 0,
            is_buy: true,
            price: "100".parse::<Decimal>().unwrap(),
            size,
            reduce_only: false,
            kind: HyperliquidExecOrderKind::Limit {
                limit: HyperliquidExecLimitParams {
                    tif: HyperliquidExecTif::Gtc,
                },
            },
            cloid: None,
        }
    }

    #[rstest]
    fn test_modify_chain_keeps_both_intents_on_rapid_modifies() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-100");
        let g0 =
            state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
        let g1 =
            state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("0.00030"));

        assert_ne!(g0, g1);
        assert!(state.has_pending_modify(&cid));
        // Front is the oldest intent
        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-0")));
        assert_eq!(
            state.pending_modify_target_qty(&cid),
            Some(Quantity::from("0.00020")),
        );
        // Both queued old legs suppress their cancel-before-accept
        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-1")));
    }

    #[rstest]
    fn test_clear_modify_generation_preserves_newer_intent() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-101");
        // Two rapid modifies queued before either acked, both against the live
        // leg v-0.
        let g0 =
            state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00030"));

        // Failure of the first modify clears only its generation; the second
        // stays, still targeting the live leg v-0.
        state.clear_modify_generation(&cid, g0);

        assert!(state.has_pending_modify(&cid));
        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-0")));
        assert_eq!(
            state.pending_modify_target_qty(&cid),
            Some(Quantity::from("0.00030")),
        );
        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
    }

    #[rstest]
    fn test_claim_front_modify_advances_next_old_id() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-102");
        // Both queued against the same stale old leg (M2 fired before M1 acked)
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00030"));

        // Promoting the first replacement claims the front and advances the
        // next intent's old leg to the replacement id.
        let claimed = state.claim_front_modify(&cid, VenueOrderId::new("v-1"));
        assert_eq!(
            claimed.map(|i| i.target_qty),
            Some(Quantity::from("0.00020"))
        );

        assert!(state.has_pending_modify(&cid));
        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-1")));
        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-1")));
        // The stale leg no longer matches once advanced
        assert!(!state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));

        // Claiming the last intent empties the chain
        let claimed2 = state.claim_front_modify(&cid, VenueOrderId::new("v-2"));
        assert_eq!(
            claimed2.map(|i| i.target_qty),
            Some(Quantity::from("0.00030"))
        );
        assert!(!state.has_pending_modify(&cid));
        assert!(state.pending_modify(&cid).is_none());
    }

    #[rstest]
    fn test_modify_chain_caps_and_evicts_oldest() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-106");
        // Queue one past the cap with no promotion or clear to drain them
        for i in 0..=MAX_PENDING_MODIFY_INTENTS {
            let voi = format!("v-{i}");
            state.mark_pending_modify(cid, VenueOrderId::new(&voi), Quantity::from("0.00020"));
        }

        // The oldest intent was evicted; the newest remains and the front
        // advanced to the second-oldest.
        assert!(!state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
        let newest = format!("v-{MAX_PENDING_MODIFY_INTENTS}");
        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new(&newest)));
        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-1")));
    }

    #[rstest]
    fn test_clear_front_modify_reparents_next_old() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-104");
        // Three rapid modifies where the first already promoted to v-1
        // (advancing the front to old=v-1); the third still holds stale v-0.
        let g_front =
            state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("0.00020"));
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00030"));

        // The front is rejected; the next intent must inherit the live leg
        // (v-1), not keep stale v-0, or CANCELED(v-1) would surface as a real
        // cancel.
        state.clear_modify_generation(&cid, g_front);

        assert!(state.has_pending_modify(&cid));
        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-1")));
        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-1")));
        assert!(!state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
    }

    #[rstest]
    fn test_clear_non_front_modify_leaves_front_old() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-105");
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
        let g_back =
            state.mark_pending_modify(cid, VenueOrderId::new("v-9"), Quantity::from("0.00030"));

        // Removing a non-front intent must not disturb the front's old leg
        state.clear_modify_generation(&cid, g_back);

        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-0")));
        assert!(!state.pending_modify_contains_old(&cid, VenueOrderId::new("v-9")));
    }

    #[rstest]
    fn test_stash_modify_request_targets_latest_intent() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-103");
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
        state.stash_modify_request(cid, sample_request(Decimal::from(1)));
        state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("0.00030"));
        state.stash_modify_request(cid, sample_request(Decimal::from(2)));

        // Front intent keeps its own request
        assert_eq!(
            state.modify_request(&cid).map(|r| r.size),
            Some(Decimal::from(1)),
        );
        // After claiming the front, the next intent's request surfaces
        state.claim_front_modify(&cid, VenueOrderId::new("v-1"));
        assert_eq!(
            state.modify_request(&cid).map(|r| r.size),
            Some(Decimal::from(2)),
        );
    }
}