// file: crates/ksp-onchain-transport-lib/src/grpc_stream.rs
// version: 6

use tonic_prost::prost::Message; // rust-rules: trait-import

const AUTO_SUBSCRIBE_PING_ID: i32 = 1;
const MAX_RECENT_UPDATE_IDENTITIES: usize = 512;
const PATH_SUBSCRIBE: &str = "/geyser.Geyser/Subscribe";

/// Safe lifecycle state of one standard Yellowstone bidirectional subscribe session.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub enum YellowstoneGrpcSubscribeState {
    /// The bidirectional stream is active and accepts request mutations.
    Active,
    /// The current stream ended and KSP is inside its bounded reconnect policy.
    Reconnecting,
    /// KSP has started a bounded graceful half-close.
    Closing,
    /// The stream ended normally, including an explicit close or a server half-close when reconnect is disabled.
    Closed,
    /// The stream terminated because a transport, protocol, decoding, backpressure or exhausted-reconnect error occurred.
    Failed,
}

/// Safe continuity and lifecycle snapshot for one standard Yellowstone Subscribe session.
///
/// Gap and duplicate counters are deliberately conservative. A continuity gap is counted only when `SubscribeReplayInfo` proves that the requested replay
/// slot is older than the endpoint's first retained slot. This proves unavailable replay coverage, not that a matching filtered update necessarily existed
/// or was lost. Duplicate updates are observed by bounded identity matching and are still delivered to the caller; KSP does not claim exactly-once delivery.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct YellowstoneGrpcSubscribeSnapshot {
    state: crate::YellowstoneGrpcSubscribeState,
    reconnect_count: u64,
    continuity_gap_count: u64,
    duplicate_update_count: u64,
    replay_attempt_count: u64,
    replay_delivery_count: u64,
    replay_coverage_unproven_count: u64,
    last_requested_from_slot: std::option::Option<u64>,
    last_observed_slot: std::option::Option<u64>,
    terminal_error_code: std::option::Option<ksp_core_lib::ErrorCode>,
}

/// Cloneable latest-value observer for one standard Yellowstone Subscribe session snapshot.
///
/// The observer exposes only the already-safe transport snapshot and keeps the internal Tokio watch channel private. Cloning the observer does not duplicate
/// the gRPC stream, request state or reconnect actor.
#[derive(Clone)]
pub struct YellowstoneGrpcSubscribeSnapshotSource {
    receiver: tokio::sync::watch::Receiver<crate::YellowstoneGrpcSubscribeSnapshot>,
}

impl crate::YellowstoneGrpcSubscribeSnapshotSource {
    fn new(receiver: tokio::sync::watch::Receiver<crate::YellowstoneGrpcSubscribeSnapshot>) -> Self {
        return Self { receiver };
    }

    /// Returns the current safe reconnect/replay snapshot without waiting for another actor transition.
    #[must_use]
    pub fn current(&self) -> crate::YellowstoneGrpcSubscribeSnapshot {
        return *self.receiver.borrow();
    }

    /// Waits for one newer safe reconnect/replay snapshot.
    ///
    /// `None` means the owning Subscribe actor dropped the latest-value publisher and no further snapshot can arrive.
    pub async fn wait_for_change(&mut self) -> std::option::Option<crate::YellowstoneGrpcSubscribeSnapshot> {
        return match self.receiver.changed().await {
            std::result::Result::Ok(()) => std::option::Option::Some(*self.receiver.borrow_and_update()),
            std::result::Result::Err(_) => std::option::Option::None,
        };
    }
}

impl std::fmt::Debug for crate::YellowstoneGrpcSubscribeSnapshotSource {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        return formatter.debug_struct("YellowstoneGrpcSubscribeSnapshotSource").field("current", &self.current()).finish();
    }
}

impl YellowstoneGrpcSubscribeSnapshot {
    const fn new(initial_from_slot: std::option::Option<u64>) -> Self {
        return Self {
            state: crate::YellowstoneGrpcSubscribeState::Active,
            reconnect_count: 0,
            continuity_gap_count: 0,
            duplicate_update_count: 0,
            replay_attempt_count: 0,
            replay_delivery_count: 0,
            replay_coverage_unproven_count: 0,
            last_requested_from_slot: initial_from_slot,
            last_observed_slot: std::option::Option::None,
            terminal_error_code: std::option::Option::None,
        };
    }

    /// Returns the current safe lifecycle state.
    #[must_use]
    pub const fn state(self) -> crate::YellowstoneGrpcSubscribeState {
        return self.state;
    }

    /// Returns the number of successful automatic stream reconnections.
    #[must_use]
    pub const fn reconnect_count(self) -> u64 {
        return self.reconnect_count;
    }

    /// Returns the number of replay discontinuities proven by `SubscribeReplayInfo` retention bounds.
    #[must_use]
    pub const fn continuity_gap_count(self) -> u64 {
        return self.continuity_gap_count;
    }

    /// Returns the number of bounded update identities observed more than once.
    ///
    /// Duplicate observations are not suppressed and this counter is not an exactly-once guarantee.
    #[must_use]
    pub const fn duplicate_update_count(self) -> u64 {
        return self.duplicate_update_count;
    }

    /// Returns the number of automatic reconnect attempts that requested a replay slot.
    #[must_use]
    pub const fn replay_attempt_count(self) -> u64 {
        return self.replay_attempt_count;
    }

    /// Returns the number of replay-bearing reconnects that delivered the requested replay boundary slot again.
    ///
    /// This is conservative delivery evidence only. It does not prove that every matching update in the replay interval was delivered and must never be
    /// interpreted as `replay_covered`.
    #[must_use]
    pub const fn replay_delivery_count(self) -> u64 {
        return self.replay_delivery_count;
    }

    /// Returns the number of successful replay-bearing reconnects whose target coverage remains unproven.
    ///
    /// The counter advances when the first post-reconnect slot-bearing update reaches or passes the requested replay boundary, because generic Transport
    /// cannot prove from that delivery alone that every matching update in the replay interval was delivered. It also advances when another reconnect starts
    /// before any slot-bearing replay material arrives. This is an explicit lack of coverage proof, not proof that a filtered event actually existed or was lost.
    #[must_use]
    pub const fn replay_coverage_unproven_count(self) -> u64 {
        return self.replay_coverage_unproven_count;
    }

    /// Returns the most recent effective `from_slot` sent by KSP, including any clamp to `SubscribeReplayInfo.first_available`.
    #[must_use]
    pub const fn last_requested_from_slot(self) -> std::option::Option<u64> {
        return self.last_requested_from_slot;
    }

    /// Returns the highest slot observed from slot-bearing standard updates.
    #[must_use]
    pub const fn last_observed_slot(self) -> std::option::Option<u64> {
        return self.last_observed_slot;
    }

    /// Returns the safe terminal KSP error code when the session failed.
    #[must_use]
    pub const fn terminal_error_code(self) -> std::option::Option<ksp_core_lib::ErrorCode> {
        return self.terminal_error_code;
    }
}

/// Standard Solana Yellowstone bidirectional `Subscribe` session layered on one KSP-owned physical gRPC channel.
///
/// Request mutations and decoded updates use bounded Tokio channels. The raw Tonic stream and upstream protobuf messages remain private to Transport.
/// When the remote stream ends, KSP can reopen it with the bounded reconnect policy from [`crate::YellowstoneGrpcSessionSettings`]. The latest accepted
/// complete request is resubmitted deterministically and `from_slot` is advanced to at least the highest observed slot. Mutations are rejected while the
/// session is reconnecting so no request can be ambiguously applied to an old or replacement stream.
pub struct SolanaYellowstoneGrpcSubscribeSession {
    endpoint_name: std::string::String,
    provider: crate::YellowstoneGrpcProviderName,
    cluster: crate::YellowstoneGrpcClusterName,
    request_state: std::sync::Arc<std::sync::Mutex<SharedRequestState>>,
    update_rx: tokio::sync::mpsc::Receiver<ksp_core_lib::Result<crate::YellowstoneSubscribeUpdate>>,
    shutdown_tx: tokio::sync::watch::Sender<std::option::Option<tokio::time::Instant>>,
    snapshot_rx: tokio::sync::watch::Receiver<crate::YellowstoneGrpcSubscribeSnapshot>,
    task: tokio::task::JoinHandle<()>,
    close_timeout: std::time::Duration,
    max_outbound_message_size_bytes: usize,
}

impl SolanaYellowstoneGrpcSubscribeSession {
    /// Returns the safe logical endpoint name.
    #[must_use]
    pub fn endpoint_name(&self) -> &str {
        return self.endpoint_name.as_str();
    }

    /// Returns the open provider descriptor without exposing endpoint credentials.
    #[must_use]
    pub const fn provider(&self) -> &crate::YellowstoneGrpcProviderName {
        return &self.provider;
    }

    /// Returns the open cluster descriptor without exposing endpoint credentials.
    #[must_use]
    pub const fn cluster(&self) -> &crate::YellowstoneGrpcClusterName {
        return &self.cluster;
    }

    /// Returns the current safe stream lifecycle state.
    #[must_use]
    pub fn state(&self) -> crate::YellowstoneGrpcSubscribeState {
        return self.snapshot_rx.borrow().state();
    }

    /// Returns the current safe reconnect/replay observability snapshot.
    #[must_use]
    pub fn snapshot(&self) -> crate::YellowstoneGrpcSubscribeSnapshot {
        return *self.snapshot_rx.borrow();
    }

    /// Returns one cloneable latest-value observer for reconnect/replay snapshot transitions.
    #[must_use]
    pub fn snapshot_source(&self) -> crate::YellowstoneGrpcSubscribeSnapshotSource {
        return crate::YellowstoneGrpcSubscribeSnapshotSource::new(self.snapshot_rx.clone());
    }

    /// Queues one complete standard Yellowstone request mutation without waiting for network dispatch.
    ///
    /// The mutation is rejected synchronously when local validation fails, the encoded request exceeds the configured outbound bound, the bounded request
    /// queue is full, or the session is not currently active. In particular, mutations are rejected during reconnect so the request state used to reopen the
    /// stream cannot race with a caller mutation.
    pub fn try_update(&self, request: &crate::YellowstoneSubscribeRequest) -> ksp_core_lib::Result<()> {
        if self.state() != crate::YellowstoneGrpcSubscribeState::Active {
            return std::result::Result::Err(subscribe_session_error(
                crate::ERROR_CODE_GRPC_SESSION_CLOSED,
                "Yellowstone subscribe session is not active for request mutation",
                self.endpoint_name.as_str(),
                self.provider.as_str(),
                self.cluster.as_str(),
            ));
        }
        let wire = match crate::yellowstone_subscribe_request_to_wire(request) {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        if wire.encoded_len() > self.max_outbound_message_size_bytes {
            return std::result::Result::Err(subscribe_session_error(
                crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW,
                "Yellowstone subscribe request exceeds the configured outbound message bound",
                self.endpoint_name.as_str(),
                self.provider.as_str(),
                self.cluster.as_str(),
            ));
        }
        let mut shared = match self.request_state.lock() {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(subscribe_session_error(
                    crate::ERROR_CODE_GRPC_SESSION_CLOSED,
                    "Yellowstone subscribe request state is unavailable",
                    self.endpoint_name.as_str(),
                    self.provider.as_str(),
                    self.cluster.as_str(),
                ));
            },
        };
        let sender = match shared.sender.as_ref() {
            std::option::Option::Some(value) => value,
            std::option::Option::None => {
                return std::result::Result::Err(subscribe_session_error(
                    crate::ERROR_CODE_GRPC_SESSION_CLOSED,
                    "Yellowstone subscribe request channel is unavailable during lifecycle transition",
                    self.endpoint_name.as_str(),
                    self.provider.as_str(),
                    self.cluster.as_str(),
                ));
            },
        };
        return match sender.try_send(wire) {
            std::result::Result::Ok(()) => {
                shared.latest_request = request.clone();
                std::result::Result::Ok(())
            },
            std::result::Result::Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => std::result::Result::Err(subscribe_session_error(
                crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW,
                "Yellowstone subscribe request queue is full",
                self.endpoint_name.as_str(),
                self.provider.as_str(),
                self.cluster.as_str(),
            )),
            std::result::Result::Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => std::result::Result::Err(subscribe_session_error(
                crate::ERROR_CODE_GRPC_SESSION_CLOSED,
                "Yellowstone subscribe request queue is closed",
                self.endpoint_name.as_str(),
                self.provider.as_str(),
                self.cluster.as_str(),
            )),
        };
    }

    /// Receives the next decoded standard Yellowstone update.
    ///
    /// Transient remote stream loss is hidden while KSP performs a bounded reconnect. A full bounded update queue applies asynchronous backpressure to the
    /// gRPC stream instead of dropping an update or terminating locally; terminal remote statuses after the reconnect budget and malformed updates still
    /// surface a safe KSP error. A normal close returns `Ok(None)`.
    pub async fn next_update(&mut self) -> ksp_core_lib::Result<std::option::Option<crate::YellowstoneSubscribeUpdate>> {
        return match self.update_rx.recv().await {
            std::option::Option::Some(std::result::Result::Ok(update)) => std::result::Result::Ok(std::option::Option::Some(update)),
            std::option::Option::Some(std::result::Result::Err(error)) => {
                clear_request_sender(&self.request_state);
                std::result::Result::Err(error)
            },
            std::option::Option::None => {
                clear_request_sender(&self.request_state);
                let snapshot = self.snapshot();
                match snapshot.state() {
                    crate::YellowstoneGrpcSubscribeState::Closed => std::result::Result::Ok(std::option::Option::None),
                    crate::YellowstoneGrpcSubscribeState::Failed => {
                        let code = snapshot.terminal_error_code().unwrap_or(crate::ERROR_CODE_GRPC_SESSION_CLOSED);
                        std::result::Result::Err(subscribe_session_error(
                            code,
                            terminal_message(code),
                            self.endpoint_name.as_str(),
                            self.provider.as_str(),
                            self.cluster.as_str(),
                        ))
                    },
                    crate::YellowstoneGrpcSubscribeState::Active
                    | crate::YellowstoneGrpcSubscribeState::Reconnecting
                    | crate::YellowstoneGrpcSubscribeState::Closing => std::result::Result::Err(subscribe_session_error(
                        crate::ERROR_CODE_GRPC_SESSION_CLOSED,
                        "Yellowstone subscribe update channel ended before a terminal state was published",
                        self.endpoint_name.as_str(),
                        self.provider.as_str(),
                        self.cluster.as_str(),
                    )),
                }
            },
        };
    }

    /// Gracefully closes the logical session and prevents any further reconnect attempt.
    pub async fn close(mut self) -> ksp_core_lib::Result<()> {
        let terminal_before_close = self.snapshot();
        clear_request_sender(&self.request_state);
        if terminal_before_close.state() == crate::YellowstoneGrpcSubscribeState::Active
            || terminal_before_close.state() == crate::YellowstoneGrpcSubscribeState::Reconnecting
        {
            let deadline = tokio::time::Instant::now() + self.close_timeout;
            self.shutdown_tx.send_replace(std::option::Option::Some(deadline));
        }
        let deadline = tokio::time::Instant::now() + self.close_timeout;
        match tokio::time::timeout_at(deadline, &mut self.task).await {
            std::result::Result::Ok(join_result) => {
                if join_result.is_err() {
                    return std::result::Result::Err(subscribe_session_error(
                        crate::ERROR_CODE_GRPC_SESSION_CLOSED,
                        "Yellowstone subscribe actor terminated unexpectedly",
                        self.endpoint_name.as_str(),
                        self.provider.as_str(),
                        self.cluster.as_str(),
                    ));
                }
            },
            std::result::Result::Err(_) => {
                self.task.abort();
                let _ = (&mut self.task).await;
                return std::result::Result::Err(subscribe_session_error(
                    crate::ERROR_CODE_TIMEOUT,
                    "Yellowstone subscribe graceful shutdown exceeded the configured close timeout",
                    self.endpoint_name.as_str(),
                    self.provider.as_str(),
                    self.cluster.as_str(),
                ));
            },
        }
        let snapshot = self.snapshot();
        return match snapshot.state() {
            crate::YellowstoneGrpcSubscribeState::Closed => std::result::Result::Ok(()),
            crate::YellowstoneGrpcSubscribeState::Failed => {
                let code = snapshot.terminal_error_code().unwrap_or(crate::ERROR_CODE_GRPC_SESSION_CLOSED);
                std::result::Result::Err(subscribe_session_error(
                    code,
                    terminal_message(code),
                    self.endpoint_name.as_str(),
                    self.provider.as_str(),
                    self.cluster.as_str(),
                ))
            },
            crate::YellowstoneGrpcSubscribeState::Active
            | crate::YellowstoneGrpcSubscribeState::Reconnecting
            | crate::YellowstoneGrpcSubscribeState::Closing => std::result::Result::Err(subscribe_session_error(
                crate::ERROR_CODE_GRPC_SESSION_CLOSED,
                "Yellowstone subscribe actor ended without a terminal lifecycle state",
                self.endpoint_name.as_str(),
                self.provider.as_str(),
                self.cluster.as_str(),
            )),
        };
    }
}

impl std::fmt::Debug for SolanaYellowstoneGrpcSubscribeSession {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let request_queue_capacity = match self.request_state.lock() {
            std::result::Result::Ok(value) => value.sender.as_ref().map(|sender| return sender.capacity()),
            std::result::Result::Err(_) => std::option::Option::None,
        };
        return formatter
            .debug_struct("SolanaYellowstoneGrpcSubscribeSession")
            .field("endpoint_name", &self.endpoint_name)
            .field("provider", &self.provider)
            .field("cluster", &self.cluster)
            .field("snapshot", &self.snapshot())
            .field("request_queue_capacity", &request_queue_capacity)
            .field("update_queue_capacity", &self.update_rx.capacity())
            .finish();
    }
}

impl Drop for SolanaYellowstoneGrpcSubscribeSession {
    fn drop(&mut self) {
        clear_request_sender(&self.request_state);
        let state = self.state();
        if state == crate::YellowstoneGrpcSubscribeState::Active || state == crate::YellowstoneGrpcSubscribeState::Reconnecting {
            self.shutdown_tx.send_replace(std::option::Option::Some(tokio::time::Instant::now() + self.close_timeout));
        }
    }
}

/// Opens one standard Yellowstone `Subscribe` stream from a validated KSP channel.
pub(crate) async fn open_yellowstone_subscribe_session(
    channel: tonic::transport::Channel,
    metadata: std::vec::Vec<crate::YellowstoneGrpcMetadataEntry>,
    settings: crate::YellowstoneGrpcSessionSettings,
    endpoint_name: std::string::String,
    provider: crate::YellowstoneGrpcProviderName,
    cluster: crate::YellowstoneGrpcClusterName,
    initial_request: crate::YellowstoneSubscribeRequest,
) -> ksp_core_lib::Result<crate::SolanaYellowstoneGrpcSubscribeSession> {
    let initial_wire = match checked_request_wire(
        &initial_request,
        settings.max_outbound_message_size_bytes(),
        endpoint_name.as_str(),
        provider.as_str(),
        cluster.as_str(),
        "initial Yellowstone subscribe request exceeds the configured outbound message bound",
    ) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let (incoming, request_tx) =
        match open_physical_subscribe_stream(channel.clone(), &metadata, &settings, endpoint_name.as_str(), provider.as_str(), cluster.as_str(), initial_wire)
            .await
        {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
    let request_state = std::sync::Arc::new(std::sync::Mutex::new(SharedRequestState {
        sender: std::option::Option::Some(request_tx),
        latest_request: initial_request.clone(),
    }));
    let (update_tx, update_rx) = tokio::sync::mpsc::channel(settings.update_channel_capacity());
    let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(std::option::Option::None::<tokio::time::Instant>);
    let initial_snapshot = crate::YellowstoneGrpcSubscribeSnapshot::new(initial_request.from_slot());
    let (snapshot_tx, snapshot_rx) = tokio::sync::watch::channel(initial_snapshot);
    let actor_request_state = request_state.clone();
    let actor_endpoint_name = endpoint_name.clone();
    let actor_provider = provider.clone();
    let actor_cluster = cluster.clone();
    let actor_settings = settings.clone();
    let close_timeout = settings.close_timeout();
    let max_outbound_message_size_bytes = settings.max_outbound_message_size_bytes();
    let task = tokio::spawn(run_subscribe_actor(
        actor_endpoint_name,
        actor_provider,
        actor_cluster,
        channel,
        metadata,
        actor_settings,
        incoming,
        actor_request_state,
        update_tx,
        shutdown_rx,
        snapshot_tx,
        initial_snapshot,
    ));
    return std::result::Result::Ok(crate::SolanaYellowstoneGrpcSubscribeSession {
        endpoint_name,
        provider,
        cluster,
        request_state,
        update_rx,
        shutdown_tx,
        snapshot_rx,
        task,
        close_timeout,
        max_outbound_message_size_bytes,
    });
}

struct SharedRequestState {
    sender: std::option::Option<tokio::sync::mpsc::Sender<yellowstone_grpc_proto::geyser::SubscribeRequest>>,
    latest_request: crate::YellowstoneSubscribeRequest,
}

struct MpscStream<T> {
    receiver: tokio::sync::mpsc::Receiver<T>,
}

impl<T> MpscStream<T> {
    const fn new(receiver: tokio::sync::mpsc::Receiver<T>) -> Self {
        return Self { receiver };
    }
}

impl<T> futures_util::Stream for MpscStream<T> {
    type Item = T;

    fn poll_next(self: std::pin::Pin<&mut Self>, context: &mut std::task::Context<'_>) -> std::task::Poll<std::option::Option<Self::Item>> {
        return self.get_mut().receiver.poll_recv(context);
    }
}

#[derive(Clone, Eq, Hash, PartialEq)]
enum UpdateIdentity {
    Account { slot: u64, pubkey: ksp_core_lib::Pubkey, write_version: u64 },
    Slot { slot: u64, status: crate::YellowstoneSlotStatus },
    Transaction { slot: u64, signature: crate::YellowstoneTransactionSignature },
    TransactionStatus { slot: u64, signature: crate::YellowstoneTransactionSignature },
    Block { slot: u64, blockhash: std::string::String },
    BlockMeta { slot: u64, blockhash: std::string::String },
    Entry { slot: u64, index: u64, hash: crate::YellowstoneHashBytes },
}

struct ContinuityTracker {
    pending_replay_from_slot: std::option::Option<u64>,
    recent_order: std::collections::VecDeque<UpdateIdentity>,
    recent_set: std::collections::HashSet<UpdateIdentity>,
}

impl ContinuityTracker {
    fn new() -> Self {
        return Self {
            pending_replay_from_slot: std::option::Option::None,
            recent_order: std::collections::VecDeque::new(),
            recent_set: std::collections::HashSet::new(),
        };
    }

    fn begin_replay(&mut self, from_slot: std::option::Option<u64>) {
        self.pending_replay_from_slot = from_slot;
        return;
    }

    fn abandon_pending_replay(&mut self, snapshot: &mut crate::YellowstoneGrpcSubscribeSnapshot) -> bool {
        if self.pending_replay_from_slot.take().is_none() {
            return false;
        }
        snapshot.replay_coverage_unproven_count = snapshot.replay_coverage_unproven_count.saturating_add(1);
        return true;
    }

    fn observe(&mut self, update: &crate::YellowstoneSubscribeUpdate, snapshot: &mut crate::YellowstoneGrpcSubscribeSnapshot) {
        if let std::option::Option::Some(slot) = update_slot(update) {
            if let std::option::Option::Some(requested) = self.pending_replay_from_slot
                && slot >= requested
            {
                if slot == requested {
                    snapshot.replay_delivery_count = snapshot.replay_delivery_count.saturating_add(1);
                }
                snapshot.replay_coverage_unproven_count = snapshot.replay_coverage_unproven_count.saturating_add(1);
                self.pending_replay_from_slot = std::option::Option::None;
            }
            snapshot.last_observed_slot = std::option::Option::Some(match snapshot.last_observed_slot {
                std::option::Option::Some(previous) => std::cmp::max(previous, slot),
                std::option::Option::None => slot,
            });
        }
        let identity = match update_identity(update) {
            std::option::Option::Some(value) => value,
            std::option::Option::None => return,
        };
        if snapshot.reconnect_count > 0 && self.recent_set.contains(&identity) {
            snapshot.duplicate_update_count = snapshot.duplicate_update_count.saturating_add(1);
        }
        if self.recent_set.insert(identity.clone()) {
            self.recent_order.push_back(identity);
            if self.recent_order.len() > MAX_RECENT_UPDATE_IDENTITIES
                && let std::option::Option::Some(oldest) = self.recent_order.pop_front()
            {
                self.recent_set.remove(&oldest);
            }
        }
    }
}

#[allow(clippy::too_many_arguments)]
async fn run_subscribe_actor(
    endpoint_name: std::string::String,
    provider: crate::YellowstoneGrpcProviderName,
    cluster: crate::YellowstoneGrpcClusterName,
    channel: tonic::transport::Channel,
    metadata: std::vec::Vec<crate::YellowstoneGrpcMetadataEntry>,
    settings: crate::YellowstoneGrpcSessionSettings,
    mut incoming: tonic::Streaming<yellowstone_grpc_proto::geyser::SubscribeUpdate>,
    request_state: std::sync::Arc<std::sync::Mutex<SharedRequestState>>,
    update_tx: tokio::sync::mpsc::Sender<ksp_core_lib::Result<crate::YellowstoneSubscribeUpdate>>,
    mut shutdown_rx: tokio::sync::watch::Receiver<std::option::Option<tokio::time::Instant>>,
    snapshot_tx: tokio::sync::watch::Sender<crate::YellowstoneGrpcSubscribeSnapshot>,
    mut snapshot: crate::YellowstoneGrpcSubscribeSnapshot,
) {
    let mut tracker = ContinuityTracker::new();
    loop {
        tokio::select! {
            shutdown_changed = shutdown_rx.changed() => {
                let deadline = shutdown_deadline(&shutdown_rx, shutdown_changed);
                snapshot.state = crate::YellowstoneGrpcSubscribeState::Closing;
                snapshot_tx.send_replace(snapshot);
                clear_request_sender(&request_state);
                finish_client_half_close(&endpoint_name, &provider, &cluster, &mut incoming, deadline, &mut snapshot, &snapshot_tx).await;
                return;
            }
            message = incoming.message() => {
                match message {
                    std::result::Result::Ok(std::option::Option::Some(wire)) => {
                        let update = match crate::yellowstone_subscribe_update_from_wire(wire) {
                            std::result::Result::Ok(value) => value,
                            std::result::Result::Err(error) => {
                                let _ = update_tx.try_send(std::result::Result::Err(error));
                                fail_actor(crate::ERROR_CODE_INVALID_RESPONSE, &request_state, &mut snapshot, &snapshot_tx);
                                return;
                            },
                        };
                        if matches!(&update, crate::YellowstoneSubscribeUpdate::Ping(_))
                            && let std::result::Result::Err(code) = send_automatic_ping(&request_state, settings.max_outbound_message_size_bytes())
                        {
                            fail_actor(code, &request_state, &mut snapshot, &snapshot_tx);
                            return;
                        }
                        tracker.observe(&update, &mut snapshot);
                        snapshot_tx.send_replace(snapshot);
                        let delivery = tokio::select! {
                            biased;
                            shutdown_changed = shutdown_rx.changed() => {
                                let deadline = shutdown_deadline(&shutdown_rx, shutdown_changed);
                                snapshot.state = crate::YellowstoneGrpcSubscribeState::Closing;
                                snapshot_tx.send_replace(snapshot);
                                clear_request_sender(&request_state);
                                finish_client_half_close(
                                    &endpoint_name,
                                    &provider,
                                    &cluster,
                                    &mut incoming,
                                    deadline,
                                    &mut snapshot,
                                    &snapshot_tx,
                                )
                                .await;
                                return;
                            }
                            result = update_tx.send(std::result::Result::Ok(update)) => result,
                        };
                        if delivery.is_err() {
                            close_actor(&request_state, &mut snapshot, &snapshot_tx);
                            return;
                        }
                    },
                    std::result::Result::Ok(std::option::Option::None) => {
                        if settings.reconnect().max_retries() == 0 {
                            close_actor(&request_state, &mut snapshot, &snapshot_tx);
                            return;
                        }
                        match reconnect_subscribe_stream(
                            &endpoint_name,
                            &provider,
                            &cluster,
                            channel.clone(),
                            &metadata,
                            &settings,
                            &request_state,
                            &mut shutdown_rx,
                            &mut snapshot,
                            &snapshot_tx,
                            &mut tracker,
                        )
                        .await
                        {
                            ReconnectOutcome::Connected(value) => incoming = value,
                            ReconnectOutcome::Shutdown => {
                                close_actor(&request_state, &mut snapshot, &snapshot_tx);
                                return;
                            },
                            ReconnectOutcome::Exhausted(error) => {
                                let _ = update_tx.try_send(std::result::Result::Err(error));
                                fail_actor(crate::ERROR_CODE_GRPC_CHANNEL_FAILED, &request_state, &mut snapshot, &snapshot_tx);
                                return;
                            },
                        }
                    },
                    std::result::Result::Err(status) => {
                        if settings.reconnect().max_retries() == 0 {
                            let error = stream_status_error("Subscribe", status, endpoint_name.as_str(), provider.as_str(), cluster.as_str());
                            let _ = update_tx.try_send(std::result::Result::Err(error));
                            fail_actor(crate::ERROR_CODE_GRPC_STATUS, &request_state, &mut snapshot, &snapshot_tx);
                            return;
                        }
                        let grpc_code = status.code().to_string();
                        ksp_logging_lib::warn!(
                            target: crate::TRACING_TARGET,
                            endpoint_name = endpoint_name.as_str(),
                            provider = provider.as_str(),
                            cluster = cluster.as_str(),
                            grpc_code = grpc_code.as_str(),
                            "Yellowstone subscribe stream ended with remote status; starting bounded reconnect"
                        );
                        match reconnect_subscribe_stream(
                            &endpoint_name,
                            &provider,
                            &cluster,
                            channel.clone(),
                            &metadata,
                            &settings,
                            &request_state,
                            &mut shutdown_rx,
                            &mut snapshot,
                            &snapshot_tx,
                            &mut tracker,
                        )
                        .await
                        {
                            ReconnectOutcome::Connected(value) => incoming = value,
                            ReconnectOutcome::Shutdown => {
                                close_actor(&request_state, &mut snapshot, &snapshot_tx);
                                return;
                            },
                            ReconnectOutcome::Exhausted(error) => {
                                let _ = update_tx.try_send(std::result::Result::Err(error));
                                fail_actor(crate::ERROR_CODE_GRPC_CHANNEL_FAILED, &request_state, &mut snapshot, &snapshot_tx);
                                return;
                            },
                        }
                    },
                }
            }
        }
    }
}

enum ReconnectOutcome {
    Connected(tonic::Streaming<yellowstone_grpc_proto::geyser::SubscribeUpdate>),
    Shutdown,
    Exhausted(ksp_core_lib::Error),
}

#[allow(clippy::too_many_arguments)]
async fn reconnect_subscribe_stream(
    endpoint_name: &str,
    provider: &crate::YellowstoneGrpcProviderName,
    cluster: &crate::YellowstoneGrpcClusterName,
    channel: tonic::transport::Channel,
    metadata: &[crate::YellowstoneGrpcMetadataEntry],
    settings: &crate::YellowstoneGrpcSessionSettings,
    request_state: &std::sync::Arc<std::sync::Mutex<SharedRequestState>>,
    shutdown_rx: &mut tokio::sync::watch::Receiver<std::option::Option<tokio::time::Instant>>,
    snapshot: &mut crate::YellowstoneGrpcSubscribeSnapshot,
    snapshot_tx: &tokio::sync::watch::Sender<crate::YellowstoneGrpcSubscribeSnapshot>,
    tracker: &mut ContinuityTracker,
) -> ReconnectOutcome {
    clear_request_sender(request_state);
    if tracker.abandon_pending_replay(snapshot) {
        snapshot_tx.send_replace(*snapshot);
    }
    snapshot.state = crate::YellowstoneGrpcSubscribeState::Reconnecting;
    snapshot.terminal_error_code = std::option::Option::None;
    snapshot_tx.send_replace(*snapshot);
    let mut backoff = settings.reconnect().initial_backoff();
    let mut gap_recorded = false;
    for attempt in 0..settings.reconnect().max_retries() {
        let sleep = tokio::time::sleep(backoff);
        tokio::pin!(sleep);
        tokio::select! {
            _ = &mut sleep => {},
            changed = shutdown_rx.changed() => {
                let _ = changed;
                return ReconnectOutcome::Shutdown;
            }
        }
        let latest_request = match clone_latest_request(request_state) {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return ReconnectOutcome::Exhausted(error),
        };
        let resume_slot = max_optional_slot(latest_request.from_slot(), snapshot.last_observed_slot);
        let mut reconnect_request = latest_request;
        let effective_from_slot = match resume_slot {
            std::option::Option::Some(requested) => {
                let first_available = replay_first_available(channel.clone(), metadata, settings.clone()).await;
                match first_available {
                    std::option::Option::Some(first_available) if first_available > requested => {
                        if !gap_recorded {
                            snapshot.continuity_gap_count = snapshot.continuity_gap_count.saturating_add(1);
                            gap_recorded = true;
                            ksp_logging_lib::warn!(
                                target: crate::TRACING_TARGET,
                                endpoint_name,
                                provider = provider.as_str(),
                                cluster = cluster.as_str(),
                                requested_from_slot = requested,
                                first_available,
                                "Yellowstone replay retention proves unavailable replay coverage; clamping reconnect from_slot"
                            );
                        }
                        std::option::Option::Some(first_available)
                    },
                    _ => std::option::Option::Some(requested),
                }
            },
            std::option::Option::None => std::option::Option::None,
        };
        reconnect_request.set_from_slot(effective_from_slot);
        snapshot.last_requested_from_slot = effective_from_slot;
        if effective_from_slot.is_some() {
            snapshot.replay_attempt_count = snapshot.replay_attempt_count.saturating_add(1);
        }
        snapshot_tx.send_replace(*snapshot);
        let wire = match checked_request_wire(
            &reconnect_request,
            settings.max_outbound_message_size_bytes(),
            endpoint_name,
            provider.as_str(),
            cluster.as_str(),
            "Yellowstone reconnect subscribe request exceeds the configured outbound message bound",
        ) {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return ReconnectOutcome::Exhausted(error),
        };
        let open = open_physical_subscribe_stream(channel.clone(), metadata, settings, endpoint_name, provider.as_str(), cluster.as_str(), wire);
        tokio::pin!(open);
        let result = tokio::select! {
            value = &mut open => std::option::Option::Some(value),
            changed = shutdown_rx.changed() => {
                let _ = changed;
                std::option::Option::None
            }
        };
        let result = match result {
            std::option::Option::Some(value) => value,
            std::option::Option::None => return ReconnectOutcome::Shutdown,
        };
        match result {
            std::result::Result::Ok((incoming, request_tx)) => {
                if let std::result::Result::Err(error) = replace_request_sender(request_state, request_tx) {
                    return ReconnectOutcome::Exhausted(error);
                }
                snapshot.reconnect_count = snapshot.reconnect_count.saturating_add(1);
                tracker.begin_replay(effective_from_slot);
                snapshot.state = crate::YellowstoneGrpcSubscribeState::Active;
                snapshot.terminal_error_code = std::option::Option::None;
                snapshot_tx.send_replace(*snapshot);
                ksp_logging_lib::info!(
                    target: crate::TRACING_TARGET,
                    endpoint_name,
                    provider = provider.as_str(),
                    cluster = cluster.as_str(),
                    reconnect_attempt = attempt.saturating_add(1),
                    replay = effective_from_slot.is_some(),
                    "reopened Yellowstone subscribe stream within bounded reconnect policy"
                );
                return ReconnectOutcome::Connected(incoming);
            },
            std::result::Result::Err(error) => {
                ksp_logging_lib::warn!(
                    target: crate::TRACING_TARGET,
                    endpoint_name,
                    provider = provider.as_str(),
                    cluster = cluster.as_str(),
                    reconnect_attempt = attempt.saturating_add(1),
                    error_code = error.code().code(),
                    "Yellowstone subscribe reconnect attempt failed"
                );
            },
        }
        backoff = std::cmp::min(backoff.saturating_mul(2), settings.reconnect().max_backoff());
    }
    return ReconnectOutcome::Exhausted(subscribe_session_error(
        crate::ERROR_CODE_GRPC_CHANNEL_FAILED,
        "Yellowstone subscribe reconnect budget is exhausted",
        endpoint_name,
        provider.as_str(),
        cluster.as_str(),
    ));
}

async fn replay_first_available(
    channel: tonic::transport::Channel,
    metadata: &[crate::YellowstoneGrpcMetadataEntry],
    settings: crate::YellowstoneGrpcSessionSettings,
) -> std::option::Option<u64> {
    let unary = crate::SolanaYellowstoneGrpcUnaryClient::new(channel, metadata.to_vec(), settings);
    return match unary.subscribe_replay_info().await {
        std::result::Result::Ok(info) => info.first_available(),
        std::result::Result::Err(error) => {
            ksp_logging_lib::warn!(
                target: crate::TRACING_TARGET,
                error_code = error.code().code(),
                "Yellowstone SubscribeReplayInfo is unavailable during reconnect; continuing without inferred retention clamp"
            );
            std::option::Option::None
        },
    };
}

#[allow(clippy::too_many_arguments)]
async fn open_physical_subscribe_stream(
    channel: tonic::transport::Channel,
    metadata: &[crate::YellowstoneGrpcMetadataEntry],
    settings: &crate::YellowstoneGrpcSessionSettings,
    endpoint_name: &str,
    provider: &str,
    cluster: &str,
    initial_wire: yellowstone_grpc_proto::geyser::SubscribeRequest,
) -> ksp_core_lib::Result<(
    tonic::Streaming<yellowstone_grpc_proto::geyser::SubscribeUpdate>,
    tokio::sync::mpsc::Sender<yellowstone_grpc_proto::geyser::SubscribeRequest>,
)> {
    let (request_tx, request_rx) = tokio::sync::mpsc::channel(settings.request_channel_capacity());
    if request_tx.try_send(initial_wire).is_err() {
        return std::result::Result::Err(subscribe_session_error(
            crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW,
            "initial Yellowstone subscribe request cannot enter the bounded request queue",
            endpoint_name,
            provider,
            cluster,
        ));
    }
    let path = match PATH_SUBSCRIBE.parse::<http::uri::PathAndQuery>() {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => {
            return std::result::Result::Err(subscribe_session_error(
                crate::ERROR_CODE_GRPC_CHANNEL_FAILED,
                "internal Yellowstone Subscribe method path is invalid",
                endpoint_name,
                provider,
                cluster,
            ));
        },
    };
    let mut grpc = tonic::client::Grpc::new(channel)
        .max_decoding_message_size(settings.max_inbound_message_size_bytes())
        .max_encoding_message_size(settings.max_outbound_message_size_bytes());
    let open_future = async {
        if grpc.ready().await.is_err() {
            return std::result::Result::Err(subscribe_session_error(
                crate::ERROR_CODE_GRPC_CHANNEL_FAILED,
                "Yellowstone gRPC channel is not ready for Subscribe dispatch",
                endpoint_name,
                provider,
                cluster,
            ));
        }
        let mut request = tonic::Request::new(MpscStream::new(request_rx));
        for entry in metadata {
            if let std::result::Result::Err(error) = entry.append_to(request.metadata_mut()) {
                return std::result::Result::Err(error);
            }
        }
        let response = match grpc
            .streaming(
                request,
                path,
                tonic_prost::ProstCodec::<yellowstone_grpc_proto::geyser::SubscribeRequest, yellowstone_grpc_proto::geyser::SubscribeUpdate>::default(),
            )
            .await
        {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(status) => {
                return std::result::Result::Err(stream_status_error("SubscribeOpen", status, endpoint_name, provider, cluster));
            },
        };
        return std::result::Result::Ok(response.into_inner());
    };
    let incoming = match tokio::time::timeout(settings.connect_timeout(), open_future).await {
        std::result::Result::Ok(std::result::Result::Ok(value)) => value,
        std::result::Result::Ok(std::result::Result::Err(error)) => return std::result::Result::Err(error),
        std::result::Result::Err(_) => {
            return std::result::Result::Err(subscribe_session_error(
                crate::ERROR_CODE_TIMEOUT,
                "Yellowstone Subscribe stream opening exceeded the configured connection timeout",
                endpoint_name,
                provider,
                cluster,
            ));
        },
    };
    return std::result::Result::Ok((incoming, request_tx));
}

async fn finish_client_half_close(
    endpoint_name: &str,
    provider: &crate::YellowstoneGrpcProviderName,
    cluster: &crate::YellowstoneGrpcClusterName,
    incoming: &mut tonic::Streaming<yellowstone_grpc_proto::geyser::SubscribeUpdate>,
    deadline: tokio::time::Instant,
    snapshot: &mut crate::YellowstoneGrpcSubscribeSnapshot,
    snapshot_tx: &tokio::sync::watch::Sender<crate::YellowstoneGrpcSubscribeSnapshot>,
) {
    loop {
        let message = tokio::time::timeout_at(deadline, incoming.message()).await;
        match message {
            std::result::Result::Ok(std::result::Result::Ok(std::option::Option::Some(_))) => {},
            std::result::Result::Ok(std::result::Result::Ok(std::option::Option::None)) => {
                snapshot.state = crate::YellowstoneGrpcSubscribeState::Closed;
                snapshot.terminal_error_code = std::option::Option::None;
                snapshot_tx.send_replace(*snapshot);
                return;
            },
            std::result::Result::Ok(std::result::Result::Err(status)) => {
                let code = status.code().to_string();
                ksp_logging_lib::warn!(
                    target: crate::TRACING_TARGET,
                    endpoint_name,
                    provider = provider.as_str(),
                    cluster = cluster.as_str(),
                    grpc_code = code.as_str(),
                    "Yellowstone subscribe endpoint returned a status during graceful shutdown"
                );
                snapshot.state = crate::YellowstoneGrpcSubscribeState::Failed;
                snapshot.terminal_error_code = std::option::Option::Some(crate::ERROR_CODE_GRPC_STATUS);
                snapshot_tx.send_replace(*snapshot);
                return;
            },
            std::result::Result::Err(_) => {
                snapshot.state = crate::YellowstoneGrpcSubscribeState::Failed;
                snapshot.terminal_error_code = std::option::Option::Some(crate::ERROR_CODE_TIMEOUT);
                snapshot_tx.send_replace(*snapshot);
                return;
            },
        }
    }
}

fn checked_request_wire(
    request: &crate::YellowstoneSubscribeRequest,
    max_outbound_message_size_bytes: usize,
    endpoint_name: &str,
    provider: &str,
    cluster: &str,
    oversized_message: &'static str,
) -> ksp_core_lib::Result<yellowstone_grpc_proto::geyser::SubscribeRequest> {
    let wire = match crate::yellowstone_subscribe_request_to_wire(request) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    if wire.encoded_len() > max_outbound_message_size_bytes {
        return std::result::Result::Err(subscribe_session_error(
            crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW,
            oversized_message,
            endpoint_name,
            provider,
            cluster,
        ));
    }
    return std::result::Result::Ok(wire);
}

fn clone_latest_request(request_state: &std::sync::Arc<std::sync::Mutex<SharedRequestState>>) -> ksp_core_lib::Result<crate::YellowstoneSubscribeRequest> {
    return match request_state.lock() {
        std::result::Result::Ok(value) => std::result::Result::Ok(value.latest_request.clone()),
        std::result::Result::Err(_) => std::result::Result::Err(ksp_core_lib::Error::new(
            crate::ERROR_CODE_GRPC_SESSION_CLOSED,
            "Yellowstone subscribe request state is unavailable during reconnect",
        )),
    };
}

fn replace_request_sender(
    request_state: &std::sync::Arc<std::sync::Mutex<SharedRequestState>>,
    sender: tokio::sync::mpsc::Sender<yellowstone_grpc_proto::geyser::SubscribeRequest>,
) -> ksp_core_lib::Result<()> {
    return match request_state.lock() {
        std::result::Result::Ok(mut value) => {
            value.sender = std::option::Option::Some(sender);
            std::result::Result::Ok(())
        },
        std::result::Result::Err(_) => std::result::Result::Err(ksp_core_lib::Error::new(
            crate::ERROR_CODE_GRPC_SESSION_CLOSED,
            "Yellowstone subscribe request state is unavailable after reconnect",
        )),
    };
}

fn clear_request_sender(request_state: &std::sync::Arc<std::sync::Mutex<SharedRequestState>>) {
    if let std::result::Result::Ok(mut value) = request_state.lock() {
        value.sender.take();
    }
}

fn send_automatic_ping(
    request_state: &std::sync::Arc<std::sync::Mutex<SharedRequestState>>,
    max_outbound_message_size_bytes: usize,
) -> std::result::Result<(), ksp_core_lib::ErrorCode> {
    let ping_wire = ping_request_wire();
    if ping_wire.encoded_len() > max_outbound_message_size_bytes {
        return std::result::Result::Err(crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW);
    }
    let shared = match request_state.lock() {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => return std::result::Result::Err(crate::ERROR_CODE_GRPC_SESSION_CLOSED),
    };
    let sender = match shared.sender.as_ref() {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return std::result::Result::Err(crate::ERROR_CODE_GRPC_SESSION_CLOSED),
    };
    return match sender.try_send(ping_wire) {
        std::result::Result::Ok(()) => std::result::Result::Ok(()),
        std::result::Result::Err(_) => std::result::Result::Err(crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW),
    };
}

fn close_actor(
    request_state: &std::sync::Arc<std::sync::Mutex<SharedRequestState>>,
    snapshot: &mut crate::YellowstoneGrpcSubscribeSnapshot,
    snapshot_tx: &tokio::sync::watch::Sender<crate::YellowstoneGrpcSubscribeSnapshot>,
) {
    clear_request_sender(request_state);
    snapshot.state = crate::YellowstoneGrpcSubscribeState::Closed;
    snapshot.terminal_error_code = std::option::Option::None;
    snapshot_tx.send_replace(*snapshot);
}

fn fail_actor(
    code: ksp_core_lib::ErrorCode,
    request_state: &std::sync::Arc<std::sync::Mutex<SharedRequestState>>,
    snapshot: &mut crate::YellowstoneGrpcSubscribeSnapshot,
    snapshot_tx: &tokio::sync::watch::Sender<crate::YellowstoneGrpcSubscribeSnapshot>,
) {
    clear_request_sender(request_state);
    snapshot.state = crate::YellowstoneGrpcSubscribeState::Failed;
    snapshot.terminal_error_code = std::option::Option::Some(code);
    snapshot_tx.send_replace(*snapshot);
}

fn update_slot(update: &crate::YellowstoneSubscribeUpdate) -> std::option::Option<u64> {
    return match update {
        crate::YellowstoneSubscribeUpdate::Account(value) => std::option::Option::Some(value.slot()),
        crate::YellowstoneSubscribeUpdate::Slot(value) => std::option::Option::Some(value.slot()),
        crate::YellowstoneSubscribeUpdate::Transaction(value) => std::option::Option::Some(value.slot()),
        crate::YellowstoneSubscribeUpdate::TransactionStatus(value) => std::option::Option::Some(value.slot()),
        crate::YellowstoneSubscribeUpdate::Block(value) => std::option::Option::Some(value.slot()),
        crate::YellowstoneSubscribeUpdate::BlockMeta(value) => std::option::Option::Some(value.slot()),
        crate::YellowstoneSubscribeUpdate::Entry(value) => std::option::Option::Some(value.entry().slot()),
        crate::YellowstoneSubscribeUpdate::Ping(_) | crate::YellowstoneSubscribeUpdate::Pong(_) => std::option::Option::None,
    };
}

fn update_identity(update: &crate::YellowstoneSubscribeUpdate) -> std::option::Option<UpdateIdentity> {
    return match update {
        crate::YellowstoneSubscribeUpdate::Account(value) => std::option::Option::Some(UpdateIdentity::Account {
            slot: value.slot(),
            pubkey: *value.account().pubkey(),
            write_version: value.account().write_version(),
        }),
        crate::YellowstoneSubscribeUpdate::Slot(value) => std::option::Option::Some(UpdateIdentity::Slot { slot: value.slot(), status: value.status() }),
        crate::YellowstoneSubscribeUpdate::Transaction(value) => {
            std::option::Option::Some(UpdateIdentity::Transaction { slot: value.slot(), signature: value.transaction().signature() })
        },
        crate::YellowstoneSubscribeUpdate::TransactionStatus(value) => {
            std::option::Option::Some(UpdateIdentity::TransactionStatus { slot: value.slot(), signature: value.signature() })
        },
        crate::YellowstoneSubscribeUpdate::Block(value) => {
            std::option::Option::Some(UpdateIdentity::Block { slot: value.slot(), blockhash: value.blockhash().to_owned() })
        },
        crate::YellowstoneSubscribeUpdate::BlockMeta(value) => {
            std::option::Option::Some(UpdateIdentity::BlockMeta { slot: value.slot(), blockhash: value.blockhash().to_owned() })
        },
        crate::YellowstoneSubscribeUpdate::Entry(value) => {
            let entry = value.entry();
            std::option::Option::Some(UpdateIdentity::Entry { slot: entry.slot(), index: entry.index(), hash: entry.hash() })
        },
        crate::YellowstoneSubscribeUpdate::Ping(_) | crate::YellowstoneSubscribeUpdate::Pong(_) => std::option::Option::None,
    };
}

fn max_optional_slot(left: std::option::Option<u64>, right: std::option::Option<u64>) -> std::option::Option<u64> {
    return match (left, right) {
        (std::option::Option::Some(left), std::option::Option::Some(right)) => std::option::Option::Some(std::cmp::max(left, right)),
        (std::option::Option::Some(value), std::option::Option::None) | (std::option::Option::None, std::option::Option::Some(value)) => {
            std::option::Option::Some(value)
        },
        (std::option::Option::None, std::option::Option::None) => std::option::Option::None,
    };
}

fn shutdown_deadline(
    shutdown_rx: &tokio::sync::watch::Receiver<std::option::Option<tokio::time::Instant>>,
    shutdown_changed: std::result::Result<(), tokio::sync::watch::error::RecvError>,
) -> tokio::time::Instant {
    return match shutdown_changed {
        std::result::Result::Ok(()) => match *shutdown_rx.borrow() {
            std::option::Option::Some(deadline) => deadline,
            std::option::Option::None => tokio::time::Instant::now(),
        },
        std::result::Result::Err(_) => tokio::time::Instant::now(),
    };
}

fn ping_request_wire() -> yellowstone_grpc_proto::geyser::SubscribeRequest {
    return yellowstone_grpc_proto::geyser::SubscribeRequest {
        ping: std::option::Option::Some(yellowstone_grpc_proto::geyser::SubscribeRequestPing { id: AUTO_SUBSCRIBE_PING_ID }),
        ..std::default::Default::default()
    };
}

fn stream_status_error(operation: &'static str, status: tonic::Status, endpoint_name: &str, provider: &str, cluster: &str) -> ksp_core_lib::Error {
    let code = status.code().to_string();
    ksp_logging_lib::warn!(
        target: crate::TRACING_TARGET,
        endpoint_name,
        provider,
        cluster,
        grpc_operation = operation,
        grpc_code = code.as_str(),
        "Yellowstone subscribe endpoint returned a gRPC status"
    );
    return subscribe_session_error(crate::ERROR_CODE_GRPC_STATUS, "Yellowstone subscribe endpoint returned a gRPC status", endpoint_name, provider, cluster)
        .with_context("grpc_operation", operation)
        .with_context("grpc_code", code);
}

fn subscribe_session_error(code: ksp_core_lib::ErrorCode, message: &'static str, endpoint_name: &str, provider: &str, cluster: &str) -> ksp_core_lib::Error {
    return ksp_core_lib::Error::new(code, message)
        .with_context("endpoint_name", endpoint_name)
        .with_context("provider", provider)
        .with_context("cluster", cluster);
}

fn terminal_message(code: ksp_core_lib::ErrorCode) -> &'static str {
    if code == crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW {
        return "Yellowstone subscribe session terminated because a bounded queue overflowed";
    }
    if code == crate::ERROR_CODE_GRPC_STATUS {
        return "Yellowstone subscribe session terminated after a remote gRPC status";
    }
    if code == crate::ERROR_CODE_GRPC_CHANNEL_FAILED {
        return "Yellowstone subscribe session exhausted its bounded reconnect policy";
    }
    if code == crate::ERROR_CODE_INVALID_RESPONSE {
        return "Yellowstone subscribe session terminated after an invalid update";
    }
    if code == crate::ERROR_CODE_TIMEOUT {
        return "Yellowstone subscribe session exceeded a configured deadline";
    }
    return "Yellowstone subscribe session is closed";
}

#[cfg(test)]
#[path = "../unit_tests/grpc_stream.rs"]
mod tests;