# msgq : Robust Redis Stream Message Queue

Robust Redis Stream based message queue with auto-claim and retry handling.

## Table of Contents

- [Features](#features)
- [Usage](#usage)
- [Design](#design)
- [Tech Stack](#tech-stack)
- [Directory Structure](#directory-structure)
- [API Reference](#api-reference)
- [History](#history)

## Features

- **Redis Stream Based**: Utilizes `XREADGROUP` for efficient message consumption.
- **Consumer Groups**: Supports scalable message processing with consumer groups.
- **Auto Group Creation**: Automatically creates consumer groups if they don't exist.
- **Reliability**: Built-in `BLOCK` and `CLAIM` mechanisms for handling idle messages.
- **Concurrency**: Processes messages concurrently using `tokio::spawn`.
- **Error Handling**: Automatic retry mechanism (configurable limit) with error callbacks.

## Usage

```rust
use std::sync::Arc;
use aok::{OK, Void};
use log::info;
use msgq::{Kv, ReadGroup};

async fn on_error(mail: Arc<Kv>, error: String) -> Void {
  info!("on_error {error} {:?}", mail);
  OK
}

async fn print_mail(mail: Arc<Kv>) -> Void {
  info!("print_mail {:?}", mail);
  OK
}

#[tokio::main]
async fn main() -> Void {
  // Initialize Redis connection (example)
  // xboot::init().await?;

  // Create a ReadGroup
  // stream: "smtp", group: "send", consumer: "test"
  // block: 1s, claim_idle: 3s, count: 2, max_retry: 3
  let read_group = ReadGroup::new("smtp", "send", "test", 1, 3, 2, 3);

  // Run the consumer loop
  read_group.run(print_mail, on_error).await?;
  OK
}
```

## Design

The `ReadGroup::run` method executes a continuous loop that ensures robust message processing:

1.  **Command Construction**: Prepares an `XREADGROUP` command with `BLOCK` (for waiting on new messages) and `CLAIM` (for recovering idle messages from other consumers) options.
2.  **Message Fetching**: Executes the command against Redis to retrieve a batch of messages.
3.  **Group Management**: Checks if the consumer group exists; if not, `auto_new` is called to create it automatically.
4.  **Concurrent Processing**: Iterates through the fetched messages and spawns a `tokio` task for each message to execute the user-provided `parse` callback.
5.  **Error Handling & Retry**:
    -   Monitors the execution of each task.
    -   If a task fails, the error is logged.
    -   The system tracks retries; if a message fails more than `max_retry` times, it is handed off to the `on_error` callback for manual intervention or logging.
6.  **Cleanup**: Successfully processed messages (or those handled by `on_error`) are acknowledged and deleted from the stream using `rm_id_li` (`XACK` + `XDEL`) to prevent reprocessing.

## Tech Stack

-   **Rust**: Core language for performance and safety.
-   **Tokio**: Asynchronous runtime for handling concurrency.
-   **Fred**: High-performance Redis client.
-   **xkv**: Redis connection and command management wrapper.

## Directory Structure

-   `src/lib.rs`: Library exports and type definitions.
-   `src/read_group.rs`: Core `ReadGroup` struct and consumer loop logic.
-   `src/auto_new.rs`: Logic for automatically creating consumer groups if missing.
-   `src/rm_id_li.rs`: Helper for acknowledging and deleting processed messages.
-   `src/parse_stream.rs`: Utilities for parsing Redis Stream responses.
-   `tests/`: Integration tests demonstrating usage patterns.

## API Reference

### `ReadGroup`

The main struct for configuring and running the consumer.

-   `new(stream, group, consumer, block_sec, claim_idle_sec, count, max_retry)`: Creates a new `ReadGroup` instance.
    -   `stream`: Redis Key for the stream.
    -   `group`: Consumer group name.
    -   `consumer`: Consumer name.
    -   `block_sec`: Seconds to block waiting for new messages.
    -   `claim_idle_sec`: Seconds before claiming idle messages from other consumers.
    -   `count`: Maximum number of messages to fetch per batch.
    -   `max_retry`: Maximum number of retries before passing to `on_error`.
-   `run(parse, on_error)`: Starts the infinite processing loop.
    -   `parse`: Async callback function to process each message.
    -   `on_error`: Async callback function to handle messages that failed retries.

### `StreamItem`

Represents a single message from the Redis Stream.

-   `id`: The unique ID of the message.
-   `retry`: The delivery count (number of times delivered).
-   `idle_ms`: Time in milliseconds the message has been idle (if claimed).
-   `kv`: The message payload as a vector of Key-Value byte pairs.

## History

In 1983, Vivek Ranadive, a 26-year-old MIT graduate, observed that while hardware components communicated via a "bus", software lacked a similar standard mechanism. He envisioned a "software bus" where applications could publish and subscribe to information without direct, rigid connections. This idea led to the creation of **The Information Bus (TIB)**, the first commercial message queue software. TIB revolutionised financial trading floors by replacing manual chalkboards with real-time digital streams, allowing different trading systems to communicate instantly. This innovation laid the groundwork for modern event-driven architectures and the message queue systems we rely on today, such as Redis Streams.