ri/c/

queue.rs

//! Copyright © 2025-2026 Wenze Wei. All Rights Reserved.
//!
//! This file is part of Ri.
//! The Ri project belongs to the Dunimd Team.
//!
//! Licensed under the Apache License, Version 2.0 (the "License");
//! You may not use this file except in compliance with the License.
//! You may obtain a copy of the License at
//!
//!     http://www.apache.org/licenses/LICENSE-2.0
//!
//! 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.

//! # Queue Module C API
//!
//! This module provides C language bindings for Ri's message queue infrastructure. The queue module
//! delivers high-performance asynchronous message processing with reliable delivery guarantees, multiple
//! queue semantics, and comprehensive routing capabilities. This C API enables C/C++ applications to
//! leverage Ri's messaging functionality for building event-driven architectures, task distribution
//! systems, and distributed processing pipelines.
//!
//! ## Module Architecture
//!
//! The queue module comprises three primary components that together provide complete messaging
//! functionality:
//!
//! - **RiQueueConfig**: Configuration container for queue parameters including queue type selection,
//!   delivery guarantees, persistence settings, and consumer group configuration. The configuration object
//!   controls queue behavior, resource allocation, and operational characteristics.
//!
//! - **RiQueueManager**: Central manager for queue lifecycle, message routing, and subscription
//!   management. The manager handles the complete messaging workflow including message production,
//!   consumption, acknowledgment, and dead-letter handling.
//!
//! - **RiQueueMessage**: Message abstraction representing individual messages in the queue system.
//!   Messages encapsulate payload data, metadata, headers, delivery properties, and routing information.
//!
//! ## Queue Types
//!
//! The queue system supports multiple queue semantics for different use cases:
//!
//! - **FIFO (First-In-First-Out) Queues**: Standard message ordering where messages are delivered
//!   in the exact order they were produced. Essential for sequential processing requirements.
//!
//! - **Priority Queues**: Messages are delivered based on priority levels rather than arrival order.
//!   High-priority messages skip ahead of lower-priority messages in the delivery sequence.
//!
//! - **Work Queues (Task Queues)**: Multiple workers compete for messages, with each message processed
//!   by exactly one worker. Enables horizontal scaling of processing capacity.
//!
//! - **Publish-Subscribe Queues**: One message published to the queue is delivered to all subscribed
//!   consumers. Enables broadcast patterns for event distribution.
//!
//! - **Delay Queues**: Messages are held invisible for a configurable delay period before becoming
//!   available for consumption. Useful for retry logic and scheduled processing.
//!
//! - **Dead Letter Queues**: Messages that fail processing after multiple attempts are moved to
//!   a separate queue for later inspection and manual handling.
//!
//! ## Delivery Guarantees
//!
//! The messaging system provides configurable delivery semantics:
//!
//! - **At-Most-Once Delivery**: Messages are delivered zero or one time. No duplication possible,
//!   but messages may be lost. Highest performance, lowest reliability.
//!
//! - **At-Least-Once Delivery**: Messages are guaranteed to be delivered at least once. Duplicates
//!   possible, but no messages lost. Requires idempotent message handlers.
//!
//! - **Exactly-Once Delivery**: Messages are delivered exactly one time. No duplication, no loss.
//!   Most complex and highest overhead. Achieved through deduplication and coordination.
//!
//! - **Transactional Delivery**: Messages are produced and consumed within database transactions.
//!   Ensures atomicity across message and data operations.
//!
//! ## Message Properties
//!
//! Each message carries comprehensive metadata:
//!
//! - **Payload**: The actual message content, stored as bytes. Can be JSON, binary, protobuf,
//!   or any custom format the application requires.
//!
//! - **Message ID**: Unique identifier for deduplication and tracking. Generated by the system
//!   or optionally specified by the producer.
//!
//! - **Correlation ID**: Application-defined identifier for relating messages to each other.
//!   Useful for request-response correlation and tracing.
//!
//! - **Timestamp**: When the message was published. Used for ordering and TTL calculations.
//!
//! - **Priority**: Message priority level (if supported by queue type). Affects delivery order.
//!
//! - **Delay**: Configurable delay before message becomes visible. Supports retry and scheduling.
//!
//! - **TTL (Time-To-Live)**: Maximum time message can remain in queue. Expired messages are
//!   removed or moved to dead letter queue.
//!
//! - **Headers**: Key-value metadata pairs for routing and processing hints. Similar to HTTP
//!   headers in purpose.
//!
//! ## Consumer Groups
//!
//! The queue system supports sophisticated consumer patterns:
//!
//! - **Shared Consumption**: Multiple consumers share messages from a queue, each message processed
//!   by one consumer. Enables load balancing across consumers.
//!
//! - **Exclusive Consumption**: One consumer receives all messages from a queue. Other consumers
//!   are blocked. Useful when ordering or stateful processing is required.
//!
//! - **Fan-Out**: Messages are replicated to multiple queues for independent consumption.
//!   Enables parallel processing pipelines from a single source.
//!
//! - **Consumer Lag Tracking**: Monitor how far behind consumers are from the message production
//!   rate. Used for capacity planning and alerting.
//!
//! ## Acknowledgment Patterns
//!
//! Message acknowledgment controls delivery guarantees:
//!
//! - **Auto-Acknowledge**: Messages are considered delivered immediately upon receipt. Simplest
//!   pattern but risks message loss on consumer failure.
//!
//! - **Manual Acknowledge**: Consumer explicitly acknowledges successful processing. Messages
//!   are only removed after successful ack. Supports reliable processing.
//!
//! - **Negative Acknowledge (Nack)**: Consumer signals processing failure, returning message
//!   to the queue for redelivery. Can optionally increase retry count.
//!
//! - **Multiple Acknowledge**: Batch multiple messages with a single acknowledgment call.
//!   Improves throughput for high-volume scenarios.
//!
//! ## Reliability Features
//! The messaging system implements comprehensive reliability mechanisms:
//!
//! - **Message Persistence**: Messages written to durable storage survive broker restarts.
//!   Configurable durability levels balance performance and reliability.
//!
//! - **Replication**: Messages copied to multiple brokers for fault tolerance. Configurable
//!   replication factor determines fault tolerance level.
//!
//! - **Checkpointing**: Consumers periodically checkpoint their progress. On restart, consumers
//!   resume from the last checkpoint rather than the beginning.
//!
//! - **Idempotent Producers**: Duplicate message detection using sequence numbers and deduplication
//!   windows. Ensures exactly-once semantics despite retries.
//!
//! ## Performance Characteristics
//!
//! Queue operations are optimized for high throughput:
//!
//! - **Message Production**: O(1) for single message, O(n) for batching
//! - **Message Consumption**: O(1) for retrieval with proper indexing
//! - **Queue Creation**: O(1) for standard queues
//! - **Throughput**: Millions of messages per second on modern hardware
//! - **Latency**: Sub-millisecond end-to-end latency for local queues
//!
//! ## Memory Management
//!
//! All C API objects use opaque pointers with manual memory management:
//!
//! - Constructor functions allocate new instances on the heap
//! - Destructor functions must be called to release memory
//! - Message payloads must be freed appropriately
//! - Queue managers coordinate resource cleanup
//!
//! ## Thread Safety
//!
//! The underlying implementations are thread-safe:
//!
//! - Concurrent message production from multiple threads supported
//! - Multiple consumers can process messages concurrently
//! - Queue operations use internal synchronization
//! - Message handling should be idempotent for concurrent processing
//!
//! ## Usage Example
//!
//! ```c
//! // Create queue configuration
//! RiQueueConfig* config = ri_queue_config_new();
//! if (config == NULL) {
//!     fprintf(stderr, "Failed to create queue config\n");
//!     return ERROR_INIT;
//! }
//!
//! // Configure queue settings
//! ri_queue_config_set_queue_type(config, QUEUE_TYPE_FIFO);
//! ri_queue_config_set_delivery_guarantee(config, DELIVERY_AT_LEAST_ONCE);
//! ri_queue_config_set_persistence_enabled(config, true);
//! ri_queue_config_set_consumer_count(config, 4);
//!
//! // Create queue manager
//! RiQueueManager* manager = ri_queue_manager_new(config);
//! if (manager == NULL) {
//!     fprintf(stderr, "Failed to create queue manager\n");
//!     ri_queue_config_free(config);
//!     return ERROR_INIT;
//! }
//!
//! // Create a message
//! RiQueueMessage* message = ri_queue_message_new();
//! if (message == NULL) {
//!     fprintf(stderr, "Failed to create message\n");
//!     ri_queue_manager_free(manager);
//!     ri_queue_config_free(config);
//!     return ERROR_INIT;
//! }
//!
//! // Configure message
//! const char* payload = "{\"event\": \"user_login\", \"user_id\": 12345}";
//! ri_queue_message_set_payload(message, payload, strlen(payload));
//! ri_queue_message_set_correlation_id(message, "login-2024-001");
//! ri_queue_message_set_priority(message, 5);
//!
//! // Set headers
//! ri_queue_message_set_header(message, "source", "auth-service");
//! ri_queue_message_set_header(message, "version", "1.0");
//!
//! // Publish message to queue
//! int result = ri_queue_manager_publish(manager, "user-events", message);
//! if (result != 0) {
//!     fprintf(stderr, "Failed to publish message: %d\n", result);
//!     ri_queue_message_free(message);
//!     ri_queue_manager_free(manager);
//!     ri_queue_config_free(config);
//!     return ERROR_PUBLISH;
//! }
//!
//! printf("Message published successfully\n");
//!
//! // Consume messages (blocking)
//! RiQueueMessage* consumed = NULL;
//! result = ri_queue_manager_consume(manager, "user-events", &consumed, 10000);
//!
//! if (result == 0 && consumed != NULL) {
//!     // Process message
//!     const char* received_payload = ri_queue_message_get_payload(consumed);
//!     size_t payload_size = ri_queue_message_get_payload_size(consumed);
//!
//!     printf("Received: %.*s\n", (int)payload_size, received_payload);
//!
//!     // Get message metadata
//!     const char* msg_id = ri_queue_message_get_id(consumed);
//!     const char* corr_id = ri_queue_message_get_correlation_id(consumed);
//!     uint64_t timestamp = ri_queue_message_get_timestamp(consumed);
//!
//!     // Process message...
//!
//!     // Acknowledge successful processing
//!     ri_queue_manager_ack(manager, consumed);
//!
//!     ri_queue_message_free(consumed);
//! } else if (result == TIMEOUT) {
//!     printf("No messages available within timeout\n");
//! } else {
//!     fprintf(stderr, "Consume error: %d\n", result);
//! }
//!
//! // Subscribe to a queue for continuous consumption
//! RiConsumerHandle* consumer = ri_queue_manager_subscribe(
//!     manager,
//!     "user-events",
//!     message_handler_callback,
//!     NULL  // user data
//! );
//!
//! if (consumer != NULL) {
//!     // Consumer runs in background
//!     printf("Consumer started, processing messages...\n");
//!
//!     // Application continues running...
//!
//!     // Stop consumer when done
//!     ri_queue_manager_unsubscribe(consumer);
//! }
//!
//! // Cleanup
//! ri_queue_message_free(message);
//! ri_queue_manager_free(manager);
//! ri_queue_config_free(config);
//! ```
//!
//! ## Message Handler Callback
//!
//! Message handlers must conform to the following signature:
//!
//! ```c
//! typedef int (*DMSQueueMessageHandler)(
//!     RiQueueManager* manager,
//!     RiQueueMessage* message,
//!     void* user_data
//! );
//! ```
//!
//! Return values:
//!
//! - 0: Success, message will be acknowledged
//! - Positive: Success with value, message acknowledged
//! - Negative: Error, message will be nacked and retried
//!
//! ## Dependencies
//!
//! This module depends on the following Ri components:
//!
//! - `crate::queue`: Rust queue module implementation
//! - `crate::prelude`: Common types and traits
//! - Async runtime for non-blocking operations
//!
//! ## Feature Flags
//!
//! The queue module is enabled by the "queue" feature flag.
//! Disable this feature to reduce binary size when messaging is not required.
//!
//! Additional features:
//!
//! - `queue-persistence`: Enable message persistence to disk
//! - `queue-rabbitmq`: Enable RabbitMQ backend support
//! - `queue-kafka`: Enable Apache Kafka backend support
//! - `queue-sqs`: Enable AWS SQS backend support

use crate::queue::{RiQueueConfig, RiQueueManager, RiQueueMessage, RiQueueStats};
use std::ffi::{c_char, c_int};
use std::sync::Arc;

c_wrapper!(CRiQueueConfig, RiQueueConfig);
c_wrapper!(CRiQueueManager, RiQueueManager);

c_constructor!(
    ri_queue_config_new,
    CRiQueueConfig,
    RiQueueConfig,
    RiQueueConfig::default()
);
c_destructor!(ri_queue_config_free, CRiQueueConfig);

#[repr(C)]
pub struct CRiQueueMessage {
    pub id: *mut c_char,
    pub payload: *mut u8,
    pub payload_len: usize,
    pub timestamp: u64,
    pub retry_count: u32,
    pub max_retries: u32,
}

#[no_mangle]
pub extern "C" fn ri_queue_message_new(payload: *const c_char, payload_len: usize) -> *mut CRiQueueMessage {
    if payload.is_null() || payload_len == 0 {
        return std::ptr::null_mut();
    }

    unsafe {
        let payload_slice = std::slice::from_raw_parts(payload as *const u8, payload_len);
        let message = RiQueueMessage::new(payload_slice.to_vec());

        let id = match std::ffi::CString::new(message.id.clone()) {
            Ok(s) => s.into_raw(),
            Err(_) => return std::ptr::null_mut(),
        };

        // Transfer the buffer ownership to C without invoking the Vec destructor.
        // Equivalent of Vec::into_raw_parts() on nightly, expressed with the
        // stable surface (as_mut_ptr + mem::forget). The C side is responsible
        // for releasing the buffer using the same global allocator.
        let mut payload_vec = message.payload.clone();
        let payload_ptr = payload_vec.as_mut_ptr();
        let payload_len_val = payload_vec.len();
        let _payload_cap = payload_vec.capacity();
        std::mem::forget(payload_vec);

        let boxed_msg = Box::new(CRiQueueMessage {
            id,
            payload: payload_ptr,
            payload_len: payload_len_val,
            timestamp: 0,
            retry_count: message.retry_count,
            max_retries: message.max_retries,
        });
        let ptr = Box::into_raw(boxed_msg);
        crate::c::register_ptr(ptr as usize);
        ptr
    }
}

#[no_mangle]
pub extern "C" fn ri_queue_message_free(msg: *mut CRiQueueMessage) {
    if msg.is_null() {
        return;
    }

    if !crate::c::unregister_ptr(msg as usize) {
        log::warn!(
            "[Ri.C] Attempted to free unregistered or already freed queue message: {:?}",
            msg
        );
        return;
    }

    unsafe {
        let msg = Box::from_raw(msg);
        if !msg.id.is_null() {
            let _ = std::ffi::CString::from_raw(msg.id);
        }
        if !msg.payload.is_null() && msg.payload_len > 0 {
            let _ = Vec::from_raw_parts(msg.payload, msg.payload_len, msg.payload_len);
        }
    }
}

#[no_mangle]
pub extern "C" fn ri_queue_message_get_id(msg: *const CRiQueueMessage) -> *const c_char {
    if msg.is_null() {
        return std::ptr::null();
    }
    unsafe { (*msg).id }
}

#[no_mangle]
pub extern "C" fn ri_queue_message_get_payload(msg: *const CRiQueueMessage, out_len: *mut usize) -> *const u8 {
    if msg.is_null() || out_len.is_null() {
        return std::ptr::null();
    }
    unsafe {
        *out_len = (*msg).payload_len;
        (*msg).payload
    }
}

#[no_mangle]
pub extern "C" fn ri_queue_manager_new() -> *mut CRiQueueManager {
    let manager = RiQueueManager::default();
    let ptr = Box::into_raw(Box::new(CRiQueueManager::new(manager)));
    crate::c::register_ptr(ptr as usize);
    ptr
}

#[no_mangle]
pub extern "C" fn ri_queue_manager_free(manager: *mut CRiQueueManager) {
    if manager.is_null() {
        return;
    }

    if !crate::c::unregister_ptr(manager as usize) {
        log::warn!(
            "[Ri.C] Attempted to free unregistered or already freed queue manager: {:?}",
            manager
        );
        return;
    }

    unsafe {
        let _ = Box::from_raw(manager);
    }
}

#[no_mangle]
pub extern "C" fn ri_queue_manager_publish(
    manager: *mut CRiQueueManager,
    queue_name: *const c_char,
    payload: *const c_char,
    payload_len: usize,
) -> c_int {
    if manager.is_null() || queue_name.is_null() || payload.is_null() {
        return -1;
    }

    unsafe {
        let queue_str = match std::ffi::CStr::from_ptr(queue_name).to_str() {
            Ok(s) => s,
            Err(_) => return -2,
        };

        let payload_slice = std::slice::from_raw_parts(payload as *const u8, payload_len);
        let message = RiQueueMessage::new(payload_slice.to_vec());

        let rt = match tokio::runtime::Runtime::new() {
            Ok(r) => r,
            Err(_) => return -3,
        };

        let result: crate::core::RiResult<Arc<dyn crate::queue::RiQueue>> = rt.block_on(async {
            (*manager).inner.create_queue(queue_str).await
        });

        match result {
            Ok(queue) => {
                let producer_result: crate::core::RiResult<Box<dyn crate::queue::RiQueueProducer>> = rt.block_on(async {
                    queue.create_producer().await
                });

                match producer_result {
                    Ok(producer) => {
                        let send_result: crate::core::RiResult<()> = rt.block_on(async {
                            producer.send(message).await
                        });
                        match send_result {
                            Ok(()) => 0,
                            Err(_) => -6,
                        }
                    }
                    Err(_) => -5,
                }
            }
            Err(_) => -4,
        }
    }
}

#[no_mangle]
pub extern "C" fn ri_queue_manager_consume(
    manager: *mut CRiQueueManager,
    queue_name: *const c_char,
    out_msg: *mut *mut CRiQueueMessage,
    _timeout_ms: u64,
) -> c_int {
    if manager.is_null() || queue_name.is_null() || out_msg.is_null() {
        return -1;
    }

    unsafe {
        let queue_str = match std::ffi::CStr::from_ptr(queue_name).to_str() {
            Ok(s) => s,
            Err(_) => return -2,
        };

        let rt = match tokio::runtime::Runtime::new() {
            Ok(r) => r,
            Err(_) => return -3,
        };

        let queue_result: Option<Arc<dyn crate::queue::RiQueue>> = rt.block_on(async {
            (*manager).inner.get_queue(queue_str).await
        });

        match queue_result {
            Some(queue) => {
                let consumer_result: crate::core::RiResult<Box<dyn crate::queue::RiQueueConsumer>> = rt.block_on(async {
                    queue.create_consumer("default_consumer").await
                });

                match consumer_result {
                    Ok(consumer) => {
                        let receive_result: crate::core::RiResult<Option<RiQueueMessage>> = rt.block_on(async {
                            consumer.receive().await
                        });

                        match receive_result {
                            Ok(Some(msg)) => {
                                let id = match std::ffi::CString::new(msg.id.clone()) {
                                    Ok(s) => s.into_raw(),
                                    Err(_) => return -7,
                                };

                                // Transfer the buffer ownership to C without invoking the Vec
                                // destructor. Stable equivalent of Vec::into_raw_parts()
                                // (as_mut_ptr + mem::forget). The C side releases the
                                // buffer through the same global allocator.
                                let mut payload_vec = msg.payload.clone();
                                let payload_ptr = payload_vec.as_mut_ptr();
                                let payload_len_val = payload_vec.len();
                                let _payload_cap = payload_vec.capacity();
                                std::mem::forget(payload_vec);

                                *out_msg = Box::into_raw(Box::new(CRiQueueMessage {
                                    id,
                                    payload: payload_ptr,
                                    payload_len: payload_len_val,
                                    timestamp: 0,
                                    retry_count: msg.retry_count,
                                    max_retries: msg.max_retries,
                                }));
                                0
                            }
                            Ok(None) => 1,
                            Err(_) => -6,
                        }
                    }
                    Err(_) => -5,
                }
            }
            None => -4,
        }
    }
}

#[repr(C)]
pub struct CRiQueueStats {
    pub queue_name: *mut c_char,
    pub message_count: u64,
    pub consumer_count: u32,
    pub producer_count: u32,
    pub processed_messages: u64,
    pub failed_messages: u64,
    pub avg_processing_time_ms: f64,
    pub total_bytes_sent: u64,
    pub total_bytes_received: u64,
    pub last_message_time: u64,
}

#[no_mangle]
pub extern "C" fn ri_queue_manager_stats(
    manager: *mut CRiQueueManager,
    queue_name: *const c_char,
    out_stats: *mut CRiQueueStats,
) -> c_int {
    if manager.is_null() || queue_name.is_null() || out_stats.is_null() {
        return -1;
    }

    unsafe {
        let queue_str = match std::ffi::CStr::from_ptr(queue_name).to_str() {
            Ok(s) => s,
            Err(_) => return -2,
        };

        let rt = match tokio::runtime::Runtime::new() {
            Ok(r) => r,
            Err(_) => return -3,
        };

        let queue_result: Option<Arc<dyn crate::queue::RiQueue>> = rt.block_on(async {
            (*manager).inner.get_queue(queue_str).await
        });

        match queue_result {
            Some(queue) => {
                let stats_result: crate::core::RiResult<RiQueueStats> = rt.block_on(async {
                    queue.get_stats().await
                });

                match stats_result {
                    Ok(stats) => {
                        let queue_name = match std::ffi::CString::new(stats.queue_name.clone()) {
                            Ok(s) => s.into_raw(),
                            Err(_) => return -5,
                        };

                        *out_stats = CRiQueueStats {
                            queue_name,
                            message_count: stats.message_count,
                            consumer_count: stats.consumer_count,
                            producer_count: stats.producer_count,
                            processed_messages: stats.processed_messages,
                            failed_messages: stats.failed_messages,
                            avg_processing_time_ms: stats.avg_processing_time_ms,
                            total_bytes_sent: stats.total_bytes_sent,
                            total_bytes_received: stats.total_bytes_received,
                            last_message_time: stats.last_message_time,
                        };
                        0
                    }
                    Err(_) => -4,
                }
            }
            None => -4,
        }
    }
}

#[no_mangle]
pub extern "C" fn ri_queue_stats_free(stats: *mut CRiQueueStats) {
    if stats.is_null() {
        return;
    }

    unsafe {
        let stats = Box::from_raw(stats);
        if !stats.queue_name.is_null() {
            let _ = std::ffi::CString::from_raw(stats.queue_name);
        }
    }
}

#[no_mangle]
pub extern "C" fn ri_queue_manager_ack(
    manager: *mut CRiQueueManager,
    queue_name: *const c_char,
    message_id: *const c_char,
) -> c_int {
    if manager.is_null() || queue_name.is_null() || message_id.is_null() {
        return -1;
    }

    unsafe {
        let queue_str = match std::ffi::CStr::from_ptr(queue_name).to_str() {
            Ok(s) => s,
            Err(_) => return -2,
        };

        let msg_id = match std::ffi::CStr::from_ptr(message_id).to_str() {
            Ok(s) => s,
            Err(_) => return -3,
        };

        let rt = match tokio::runtime::Runtime::new() {
            Ok(r) => r,
            Err(_) => return -4,
        };

        let queue_result: Option<Arc<dyn crate::queue::RiQueue>> = rt.block_on(async {
            (*manager).inner.get_queue(queue_str).await
        });

        match queue_result {
            Some(queue) => {
                let consumer_result: crate::core::RiResult<Box<dyn crate::queue::RiQueueConsumer>> = rt.block_on(async {
                    queue.create_consumer("default_consumer").await
                });

                match consumer_result {
                    Ok(consumer) => {
                        let ack_result: crate::core::RiResult<()> = rt.block_on(async {
                            consumer.ack(msg_id).await
                        });
                        match ack_result {
                            Ok(()) => 0,
                            Err(_) => -7,
                        }
                    }
                    Err(_) => -6,
                }
            }
            None => -5,
        }
    }
}

#[no_mangle]
pub extern "C" fn ri_queue_manager_nack(
    manager: *mut CRiQueueManager,
    queue_name: *const c_char,
    message_id: *const c_char,
) -> c_int {
    if manager.is_null() || queue_name.is_null() || message_id.is_null() {
        return -1;
    }

    unsafe {
        let queue_str = match std::ffi::CStr::from_ptr(queue_name).to_str() {
            Ok(s) => s,
            Err(_) => return -2,
        };

        let msg_id = match std::ffi::CStr::from_ptr(message_id).to_str() {
            Ok(s) => s,
            Err(_) => return -3,
        };

        let rt = match tokio::runtime::Runtime::new() {
            Ok(r) => r,
            Err(_) => return -4,
        };

        let queue_result: Option<Arc<dyn crate::queue::RiQueue>> = rt.block_on(async {
            (*manager).inner.get_queue(queue_str).await
        });

        match queue_result {
            Some(queue) => {
                let consumer_result: crate::core::RiResult<Box<dyn crate::queue::RiQueueConsumer>> = rt.block_on(async {
                    queue.create_consumer("default_consumer").await
                });

                match consumer_result {
                    Ok(consumer) => {
                        let nack_result: crate::core::RiResult<()> = rt.block_on(async {
                            consumer.nack(msg_id).await
                        });
                        match nack_result {
                            Ok(()) => 0,
                            Err(_) => -7,
                        }
                    }
                    Err(_) => -6,
                }
            }
            None => -5,
        }
    }
}