v_queue 0.2.9

# API Reference

Complete HTTP API documentation for V-Queue server.

**Important Note**: The v-queue-server provides a **consumer-only HTTP API**. There are no HTTP endpoints for producing/pushing messages. To produce messages, you must use the v_queue library directly in your Rust application or via JNI bindings for Java.

## Base URL

```
http://localhost:9093/api/v1
```

## Authentication

Most endpoints require HTTP Basic Authentication (when `auth_enabled = true`).

```bash
curl -u username:password http://localhost:9093/api/v1/queues
```

Or with Authorization header:

```bash
curl -H "Authorization: Basic <base64_credentials>" http://localhost:9093/api/v1/queues
```

## Endpoints

### Health Check

#### GET /health

Check server health status.

**Authentication**: Not required

**Response**:
```json
{
  "status": "ok",
  "version": "0.1.0"
}
```

**Status Codes**:
- `200 OK` - Server is healthy

**Example**:
```bash
curl http://localhost:9093/health
```

---

### List Queues

#### GET /api/v1/queues

List all available queues.

**Authentication**: Required

**Response**:
```json
{
  "queues": ["events", "logs", "metrics"]
}
```

**Status Codes**:
- `200 OK` - Success
- `401 Unauthorized` - Authentication failed

**Example**:
```bash
curl -u admin:password http://localhost:9093/api/v1/queues
```

---

### Get Queue Information

#### GET /api/v1/queues/{queue}

Get detailed information about a specific queue.

**Authentication**: Required

**Path Parameters**:
- `queue` (string) - Queue name

**Response**:
```json
{
  "name": "events",
  "id": 0,
  "count_pushed": 42,
  "is_ready": true
}
```

**Response Fields**:
- `name` - Queue name
- `id` - Current active partition ID
- `count_pushed` - Number of messages in current partition
- `is_ready` - Queue operational status

**Status Codes**:
- `200 OK` - Success
- `401 Unauthorized` - Authentication failed
- `404 Not Found` - Queue doesn't exist
- `500 Internal Server Error` - Server error

**Example**:
```bash
curl -u admin:password http://localhost:9093/api/v1/queues/events
```

---

### List Consumers

#### GET /api/v1/queues/{queue}/consumers

List all consumers for a specific queue.

**Authentication**: Required

**Path Parameters**:
- `queue` (string) - Queue name

**Response**:
```json
{
  "consumers": ["consumer1", "consumer2", "web-app"]
}
```

**Status Codes**:
- `200 OK` - Success
- `401 Unauthorized` - Authentication failed
- `404 Not Found` - Queue doesn't exist

**Example**:
```bash
curl -u admin:password http://localhost:9093/api/v1/queues/events/consumers
```

---

### Consume Messages

#### GET /api/v1/queues/{queue}/consumers/{consumer}/messages

Retrieve messages from a queue for a specific consumer.

**Authentication**: Required

**Path Parameters**:
- `queue` (string) - Queue name
- `consumer` (string) - Consumer name (created automatically if doesn't exist)

**Query Parameters**:
- `timeout_ms` (integer, optional) - Long polling timeout in milliseconds (default: 1000)
  - `0` - Return immediately if no messages
  - `> 0` - Wait up to N milliseconds for messages
- `max_messages` (integer, optional) - Maximum messages to return (default: 1)
  - Range: 1-10000
  - Note: The `default_batch_size` config setting is not currently used for this parameter

**Response**:
```json
{
  "messages": [
    {
      "offset": 0,
      "msg_type": "string",
      "value": "Hello World"
    },
    {
      "offset": 42,
      "msg_type": "object",
      "value": {"event": "user_login", "user_id": 123}
    },
    {
      "offset": 98,
      "msg_type": "object",
      "value": "[Binary Data]",
      "raw_bytes": "SGVsbG8gV29ybGQ="
    }
  ]
}
```

**Response Fields**:

Message object:
- `offset` (integer) - Message offset in queue
- `msg_type` (string) - Message type: `"string"` or `"object"`
- `value` - Message content:
  - String messages: UTF-8 string
  - Object messages: JSON object or `"[Binary Data]"`
- `raw_bytes` (string, optional) - Base64-encoded raw bytes (for binary data)

**Status Codes**:
- `200 OK` - Success (may return empty array if no messages)
- `401 Unauthorized` - Authentication failed
- `404 Not Found` - Queue doesn't exist
- `500 Internal Server Error` - Server error

**Behavior**:

1. **First Request**: Consumer created with offset 0
2. **Subsequent Requests**: Reads from last position
3. **Offset NOT Updated**: Offset only updated by commit
4. **Re-reading**: Same messages returned until committed

**Examples**:

Consume immediately:
```bash
curl -u admin:password \
  "http://localhost:9093/api/v1/queues/events/consumers/my-app/messages"
```

With timeout (long polling):
```bash
curl -u admin:password \
  "http://localhost:9093/api/v1/queues/events/consumers/my-app/messages?timeout_ms=30000"
```

Limit messages:
```bash
curl -u admin:password \
  "http://localhost:9093/api/v1/queues/events/consumers/my-app/messages?max_messages=10"
```

Combined:
```bash
curl -u admin:password \
  "http://localhost:9093/api/v1/queues/events/consumers/my-app/messages?timeout_ms=5000&max_messages=100"
```

---

### Commit Consumer Position

#### POST /api/v1/queues/{queue}/consumers/{consumer}/commit

Commit the current consumer position, marking messages as processed.

**Authentication**: Required

**Path Parameters**:
- `queue` (string) - Queue name
- `consumer` (string) - Consumer name

**Request Body**: None required

**Response**: Empty body with HTTP status code only

**Status Codes**:
- `200 OK` - Successfully committed
- `401 Unauthorized` - Authentication failed
- `404 Not Found` - Queue or consumer doesn't exist
- `500 Internal Server Error` - Server error

**Behavior**:

1. Saves current read position to disk
2. Subsequent consume requests start from committed position
3. Prevents re-reading already processed messages

**Example**:
```bash
curl -X POST -u admin:password \
  http://localhost:9093/api/v1/queues/events/consumers/my-app/commit
```

---

## Message Types

### String Messages

Text messages stored as UTF-8 strings.

**Type**: `"string"`

**Example**:
```json
{
  "offset": 0,
  "msg_type": "string",
  "value": "Hello World"
}
```

### Object/Binary Messages

Structured data or binary content.

**Type**: `"object"`

**JSON Object**:
```json
{
  "offset": 42,
  "msg_type": "object",
  "value": {
    "event": "user_login",
    "timestamp": "2024-01-15T10:30:00Z"
  }
}
```

**Binary Data**:
```json
{
  "offset": 98,
  "msg_type": "object",
  "value": "[Binary Data]",
  "raw_bytes": "SGVsbG8gV29ybGQ="
}
```

## Consumer Workflow

Typical consumer workflow:

```
1. GET /messages?timeout_ms=30000  → Receive messages
2. Process messages                → Application logic
3. POST /commit                    → Mark as processed
4. Repeat from step 1
```

## Error Handling

### Error Response Format

```json
{
  "error": "Error description"
}
```

### Common Error Codes

#### 400 Bad Request

Invalid request parameters.

**Example**:
```json
{
  "error": "Invalid max_messages parameter"
}
```

#### 401 Unauthorized

Authentication required or invalid credentials.

**Example**:
```json
{
  "error": "Authentication required"
}
```

#### 404 Not Found

Queue or consumer doesn't exist.

**Example**:
```json
{
  "error": "Queue not found: unknown-queue"
}
```

#### 500 Internal Server Error

Server-side error.

**Example**:
```json
{
  "error": "Failed to open queue"
}
```

## Usage Examples

### Python

```python
import requests
from requests.auth import HTTPBasicAuth

BASE_URL = "http://localhost:9093/api/v1"
auth = HTTPBasicAuth("admin", "password")

# List queues
response = requests.get(f"{BASE_URL}/queues", auth=auth)
queues = response.json()["queues"]

# Consume messages
response = requests.get(
    f"{BASE_URL}/queues/events/consumers/my-app/messages",
    params={"timeout_ms": 30000, "max_messages": 100},
    auth=auth
)
messages = response.json()["messages"]

# Process messages
for msg in messages:
    print(f"Offset: {msg['offset']}, Value: {msg['value']}")

# Commit
requests.post(
    f"{BASE_URL}/queues/events/consumers/my-app/commit",
    auth=auth
)
```

### JavaScript (Node.js)

```javascript
const axios = require('axios');

const client = axios.create({
  baseURL: 'http://localhost:9093/api/v1',
  auth: {
    username: 'admin',
    password: 'password'
  }
});

async function consumeMessages() {
  // Consume with timeout
  const response = await client.get(
    '/queues/events/consumers/my-app/messages',
    {
      params: {
        timeout_ms: 30000,
        max_messages: 100
      }
    }
  );

  const messages = response.data.messages;

  // Process messages
  for (const msg of messages) {
    console.log(`Offset: ${msg.offset}, Value: ${msg.value}`);
  }

  // Commit
  await client.post('/queues/events/consumers/my-app/commit');
}

consumeMessages().catch(console.error);
```

### Rust

```rust
use reqwest::blocking::Client;
use serde::{Deserialize, Serialize};

#[derive(Deserialize)]
struct MessagesResponse {
    messages: Vec<Message>,
}

#[derive(Deserialize)]
struct Message {
    offset: u64,
    msg_type: String,
    value: serde_json::Value,
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::new();

    // Consume messages
    let response = client
        .get("http://localhost:9093/api/v1/queues/events/consumers/my-app/messages")
        .basic_auth("admin", Some("password"))
        .query(&[("timeout_ms", "30000"), ("max_messages", "100")])
        .send()?
        .json::<MessagesResponse>()?;

    // Process messages
    for msg in response.messages {
        println!("Offset: {}, Value: {}", msg.offset, msg.value);
    }

    // Commit
    client
        .post("http://localhost:9093/api/v1/queues/events/consumers/my-app/commit")
        .basic_auth("admin", Some("password"))
        .send()?;

    Ok(())
}
```

### cURL

```bash
#!/bin/bash

BASE_URL="http://localhost:9093/api/v1"
AUTH="admin:password"

# Consume messages
MESSAGES=$(curl -s -u "$AUTH" \
  "$BASE_URL/queues/events/consumers/my-app/messages?timeout_ms=30000&max_messages=100")

echo "$MESSAGES" | jq '.messages[]'

# Commit
curl -s -X POST -u "$AUTH" \
  "$BASE_URL/queues/events/consumers/my-app/commit"
```

## Rate Limits

Currently, no rate limits are enforced. Consider implementing rate limiting at the reverse proxy level if needed.

## Best Practices

### Consumer Naming

Use descriptive, unique consumer names:

- ✅ Good: `analytics-service`, `email-processor`, `audit-logger`
- ❌ Bad: `consumer1`, `test`, `temp`

### Timeout Values

Choose appropriate timeouts:

- **Short-lived jobs**: `timeout_ms=5000` (5 seconds)
- **Background processing**: `timeout_ms=30000` (30 seconds)
- **Batch processing**: `timeout_ms=60000` (60 seconds)
- **Real-time**: `timeout_ms=0` (no waiting)

### Batch Sizes

Balance throughput and latency:

- **Low latency**: `max_messages=10`
- **Balanced**: `max_messages=100`
- **High throughput**: `max_messages=1000`

Note: Default is `max_messages=1`. Specify a higher value for better throughput.

### Error Handling

Always handle errors:

```python
try:
    response = requests.get(url, timeout=35)  # HTTP timeout > server timeout
    response.raise_for_status()
    messages = response.json()["messages"]
except requests.Timeout:
    print("Request timed out")
except requests.HTTPError as e:
    print(f"HTTP error: {e}")
except Exception as e:
    print(f"Error: {e}")
```

### Commit Strategy

**Option 1 - Commit After Each Batch**:
```python
messages = consume()
process(messages)
commit()  # Safe, but more commits
```

**Option 2 - Commit Periodically**:
```python
for i in range(10):
    messages = consume()
    process(messages)
commit()  # Fewer commits, risk of reprocessing
```

**Option 3 - Commit After Successful Processing**:
```python
try:
    messages = consume()
    process(messages)
    commit()  # Only commit if successful
except Exception:
    # Don't commit, will retry
    pass
```

## Next Steps

- [Authentication Guide](06-authentication.md)
- [Client Examples](07-client-examples.md)
- [Performance Tuning](08-performance.md)