Metadata-Version: 2.5
Name: csrd-message
Version: 0.8.0
Summary: Messaging abstractions for publisher/consumer workflows with optional RabbitMQ, Redis Streams and Kafka extras
Project-URL: Repository, https://github.com/csrd-api/fastapi-common
Project-URL: Documentation, https://github.com/csrd-api/fastapi-common/tree/main/packages/message
Project-URL: Changelog, https://github.com/csrd-api/fastapi-common/blob/main/CHANGELOG.md
License: MIT
Requires-Python: >=3.12
Provides-Extra: all
Requires-Dist: aio-pika<10,>=9; extra == 'all'
Requires-Dist: aiokafka<1,>=0.11; extra == 'all'
Requires-Dist: redis<8,>=5; extra == 'all'
Provides-Extra: kafka
Requires-Dist: aiokafka<1,>=0.11; extra == 'kafka'
Provides-Extra: rabbit
Requires-Dist: aio-pika<10,>=9; extra == 'rabbit'
Provides-Extra: redis
Requires-Dist: redis<8,>=5; extra == 'redis'
Description-Content-Type: text/markdown

# csrd.message

Transport-agnostic message abstractions for publish/consume workflows, with
optional RabbitMQ and Redis Streams adapters.

## Why this package exists

- Provide a stable API for messaging behavior independent of broker choice
- Enable optional broker installs with extras (`rabbit`, `redis`, `kafka`)
- Keep service code portable across RabbitMQ, Redis Streams and Kafka backends

## Installation

```bash
# Core abstractions only
uv add csrd-message

# RabbitMQ support
uv add 'csrd-message[rabbit]'

# Redis Streams support (redis-py >= 5, redis.asyncio)
uv add 'csrd-message[redis]'

# Kafka support (dependency only; no adapter yet)
uv add 'csrd-message[kafka]'

# Everything
uv add 'csrd-message[all]'
```

## Important extras note

Python extras are part of the same PyPI project, not a separate project.

- `csrd-message` and `csrd-message[redis]` resolve to the same package name (`csrd-message`)
- The extra only adds optional dependencies declared by this package
- Extras do not create new package names on PyPI

## Core API surface

- `Message` - immutable message envelope
- `MessagePublisher` - protocol for publishing messages
- `MessageConsumer` - protocol for consuming messages (`consume(message) -> Acknowledgement`)
- `Acknowledgement` - `ACK` / `RETRY` / `REJECT`
- `encode_message` / `decode_message` (and `message_to_dict` / `message_from_dict`) -
  the canonical JSON envelope (see [Wire encoding](#wire-encoding))

```python
from csrd.message import Acknowledgement, Message, MessagePublisher


async def send(publisher: MessagePublisher) -> None:
    await publisher.publish(Message(topic="orders.created", payload={"order_id": "o-1"}))
```

## Redis Streams adapter (`csrd-message[redis]`)

```python
from csrd.message.redis import RedisConsumer, RedisLifespan, RedisPublisher, StreamConfig
```

At-least-once queueing over Redis Streams consumer groups. Use it when a
csrd service already runs Redis and needs a job/event queue without adding
RabbitMQ. See `docs/MESSAGING_AND_WORKERS.md` for how it relates to Celery.

### Publisher

```python
RedisPublisher(
    redis,                    # redis.asyncio.Redis
    *,
    stream_prefix="",         # stream key = f"{stream_prefix}{message.topic}"
    maxlen=10_000,            # XADD MAXLEN; None = never trim
    approximate=True,         # MAXLEN ~ (recommended)
)
await publisher.publish(message)               # MessagePublisher protocol
entry_id = await publisher.publish_entry(message)  # same, returns the stream entry id
```

### Consumer

```python
RedisConsumer(
    redis, topic, group, consumer,   # consumer: any MessageConsumer
    *,
    consumer_name=None,      # default: socket.gethostname(); must be stable across restarts
    stream_prefix="",
    concurrency=10,          # max messages processed at once
    block_ms=5_000,          # XREADGROUP BLOCK
    batch_size=10,           # COUNT for XREADGROUP / XAUTOCLAIM
    claim_idle_ms=60_000,    # XAUTOCLAIM min idle time (also the RETRY delay)
    claim_interval_ms=30_000,  # how often to XAUTOCLAIM; must be < claim_idle_ms
    max_deliveries=5,        # dead-letter after this many deliveries; None = unlimited
    dead_letter_suffix=":dead",
    dead_letter_maxlen=None,
    group_start_id="0",      # XGROUP CREATE id; "0" also consumes entries published before the group existed
    shutdown_timeout=30.0,   # seconds stop() waits for in-flight messages
)
```

- `await consumer.run()` consumes until `await consumer.stop()` (use from a
  standalone worker process; call `stop()` from a signal handler).
- `consumer.start()` runs it as a background task; `await consumer.stop()` stops it.
- Connection errors are retried with exponential backoff (0.5 s up to 30 s);
  the group is recreated if it disappears (`NOGROUP`).

Processing loop:

1. `XGROUP CREATE <stream> <group> <group_start_id> MKSTREAM`. `BUSYGROUP` is ignored.
2. **Restart recovery**: `XREADGROUP GROUP <group> <consumer_name> STREAMS <stream> 0`
   pages through this consumer's own pending entries and dispatches them again.
3. Every `claim_interval_ms`: `XCLAIM ... JUSTID` the in-flight ids to self.
   This resets their idle time, so other consumers never steal a message that
   is still being processed. Then `XAUTOCLAIM` entries idle longer than
   `claim_idle_ms`: those left by crashed consumers and those that got `RETRY`.
4. `XREADGROUP ... BLOCK <block_ms> COUNT <n> STREAMS <stream> >`, where `n`
   is never more than the free concurrency slots.

Entry ids already in flight in this process are never dispatched a second
time, even when a reclaim returns them.

### Acknowledgement mapping

| Outcome of `consume(message)` | Redis action |
|---|---|
| `Acknowledgement.ACK` | `XACK` |
| `Acknowledgement.REJECT` | `XADD <stream><dead_letter_suffix>`, then `XACK` |
| `Acknowledgement.RETRY` | none; entry stays pending, re-delivered by `XAUTOCLAIM` after `claim_idle_ms` |
| raises an exception | logged, treated as `RETRY` |
| delivery count > `max_deliveries` | dead-lettered (`reason=max_deliveries`) without calling `consume` |
| entry is not valid JSON or lacks the `message` field | dead-lettered immediately (`reason=undecodable`) |
| entry was trimmed while pending | `XACK` (nothing to process) |
| cancelled at shutdown (after `shutdown_timeout`) | stays pending, recovered on restart |

The exception and poison handling mirrors the Rabbit adapter. There, a handler
error is dead-lettered to a TTL queue and re-delivered, and a message rejected
5 or more times is dropped. The Redis adapter dead-letters it rather than
dropping it.

Delivery is **at least once**: `consume` must be idempotent, keyed for
example on `message.message_id`.

### Lifespan (FastAPI)

```python
from csrd.lifespan import lifespan_stack
from csrd.message.redis import RedisLifespan, StreamConfig

redis_lifespan = RedisLifespan(
    "redis://:password@redis:6379/0",      # or client=existing_redis (not closed on shutdown)
    streams=[StreamConfig(topic="jobs.render", group="render-workers", consumer=RenderConsumer())],
    stream_prefix="",
    maxlen=10_000,
    consumer_name=None,                    # default for StreamConfig.consumer_name
    state_key="redis_streams",
)
app = FastAPI(lifespan=lifespan_stack(db_lifespan, redis_lifespan))

# in a dependency:
state = request.app.state.redis_streams        # RedisStreamsState
await state.publisher.publish(Message(topic="jobs.render", payload={...}))
```

`StreamConfig` has the same fields and defaults as the `RedisConsumer` keyword
arguments. If you omit `streams`, the lifespan is publish-only. The state is
stored under `redis_streams`, not `redis`, so it does not clash with a
service's own `app.state.redis` client. On shutdown every consumer is stopped
gracefully, and the client is closed only if the lifespan created it.

### Wire encoding

This section is the cross-language contract.

- **Stream key**: `<stream_prefix><topic>`. With no prefix, the stream key is
  the topic itself, for example `jobs.render`.
- **Entry fields**: exactly one field, `message`, whose value is a UTF-8 JSON
  object (the envelope). Entries without it are dead-lettered.
- **Consumer groups**: a normal Redis consumer group. Any language may join
  the same group or read the stream with its own group.
- **Dead-letter stream**: `<stream><dead_letter_suffix>` (default
  `jobs.render:dead`). Its fields are `message` (the original envelope,
  unchanged), `source_stream`, `source_id` (the original entry id), `group`,
  `consumer`, `reason` (`rejected` | `max_deliveries` | `undecodable`) and,
  when known, `deliveries`.

Envelope (JSON Schema):

```json
{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "csrd.message.Message",
  "type": "object",
  "required": ["topic", "payload"],
  "properties": {
    "topic":      {"type": "string", "minLength": 1},
    "payload":    {"type": "object"},
    "key":        {"type": ["string", "null"], "default": null},
    "headers":    {"type": "object", "additionalProperties": {"type": "string"}, "default": {}},
    "message_id": {"type": ["string", "null"], "default": null},
    "timestamp":  {"type": "string", "format": "date-time",
                   "description": "ISO-8601; producers SHOULD include a UTC offset. Naive values are read as UTC. Defaults to the time of decoding."}
  }
}
```

Example entry, as created by `XADD jobs.render MAXLEN ~ 10000 * message '<json>'`:

```json
{"topic":"jobs.render","payload":{"card_id":"c-1"},"key":null,"headers":{"trace-id":"4bf9"},"message_id":"9a3e...","timestamp":"2026-10-16T12:00:00.123456+00:00"}
```

The Python encoder writes compact JSON and refuses payloads that the standard
`json` module cannot serialise. Convert datetimes, UUIDs and similar values to
strings yourself. `topic` may be omitted by foreign producers: the consumer
then uses the topic it is subscribed to. Unknown envelope keys are ignored.

### Observability

The Redis adapter logs through the standard library, like the Rabbit adapter,
using the logger names `csrd.message.redis.*`. Records carry `extra` fields
`stream`, `entry_id`, `group` and, once decoded, `message_id`, so
`csrd-logging` JSON formatters pick them up when that package is installed.
`csrd-message` does not import `csrd-logging`, `csrd-context` or
`csrd-metrics`. To bind context or count outcomes, wrap your
`MessageConsumer`.

## RabbitMQ adapter (`csrd-message[rabbit]`)

`from csrd.message.rabbit import RabbitPublisher, RabbitConsumer, RabbitMessageHandler, RabbitLifespan, QueueConfig, AMQPState`

See `docs/PACKAGE_CAPABILITIES.md` and `docs/MESSAGING_AND_WORKERS.md`.
