cdk-axum 0.17.1

use cdk::subscription::Params;
use cdk::ws::{WsResponseResult, WsSubscribeResponse};

use super::{WsContext, WsError, MAX_FILTERS_PER_SUBSCRIPTION, MAX_SUBSCRIPTIONS_PER_CONNECTION};

/// The `handle` method is called when a client sends a subscription request
pub(crate) async fn handle(
    context: &mut WsContext,
    params: Params,
) -> Result<WsResponseResult, WsError> {
    let sub_id = params.id.clone();
    if context.subscriptions.contains_key(&sub_id) {
        // Subscription ID already exits. Returns an error instead of
        // replacing the other subscription or avoiding it.
        return Err(WsError::InvalidParams);
    }

    if context.subscriptions.len() >= MAX_SUBSCRIPTIONS_PER_CONNECTION {
        tracing::warn!(
            "WebSocket subscription request exceeds per-connection limit: {} >= {}",
            context.subscriptions.len(),
            MAX_SUBSCRIPTIONS_PER_CONNECTION
        );
        return Err(WsError::InvalidParams);
    }

    let max_filters = MAX_FILTERS_PER_SUBSCRIPTION;
    if params.filters.len() > max_filters {
        tracing::warn!(
            "WebSocket subscription request exceeds max filters limit: {} > {}",
            params.filters.len(),
            max_filters
        );
        return Err(WsError::InvalidParams);
    }

    let mut subscription = context
        .state
        .mint
        .pubsub_manager()
        .subscribe(params)
        .map_err(|_| WsError::ParseError)?;

    let publisher = context.publisher.clone();
    let sub_id_for_sender = sub_id.clone();
    context.subscriptions.insert(
        sub_id.clone(),
        tokio::spawn(async move {
            while let Some(response) = subscription.recv().await {
                let _ = publisher.try_send((sub_id_for_sender.clone(), response.into_inner()));
            }
        }),
    );
    Ok(WsSubscribeResponse {
        status: "OK".to_string(),
        sub_id,
    }
    .into())
}