# Kanject.Core.Queue

A typed SQS layer: publish a message with `IQueueManager.EnqueueAsync`, consume a batch by marking a `partial` class `[QueueConsumer]` — the source generator makes it an `AbstractQueueConsumer<T>` with the receive loop, acknowledgement, retries, and dead-letter handling wired in. Free under the Community licence.

## Install

```bash
dotnet add package Kanject.Core.Queue.Provider.AwsSqs
```

## Define the message

```csharp
using Kanject.Core.Queue.Provider.AwsSqs.Annotations.Attributes;

// The message. [QueueMessage] names the queue it belongs to.
[QueueMessage("order-events")]
public record struct OrderPlaced(Guid OrderId, string CustomerId);
```

## Produce

Inject `IQueueManager` and enqueue the typed message — serialization and the target queue are resolved from the `[QueueMessage]` attribute:

```csharp
// Produce: inject IQueueManager and enqueue a typed message.
public class CheckoutService(IQueueManager queue)
{
    public Task PlaceAsync(Guid orderId, string customerId)
        => queue.EnqueueAsync(new OrderPlaced(orderId, customerId));
}
```

## Consume

A consumer is a `partial` class with `[QueueConsumer]`. The generator makes it inherit `AbstractQueueConsumer<T>`; you override the **batched** `ConsumeAsync`, `AcknowledgeAsync` the messages you handle, and push the rest onto `Response.BatchItemFailures` so SQS redelivers only those:

```csharp
using Kanject.Core.Queue.Provider.AwsSqs.Annotations.Attributes;

// Consume: a partial class the generator makes inherit
// AbstractQueueConsumer<OrderPlaced>. You override ConsumeAsync (batched)
// and acknowledge each message you successfully handle.
[QueueConsumer(
    QueueName      = "order-events",
    QueueNamespace = "beanandbark",
    Message        = typeof(OrderPlaced))]
public partial class OrderPlacedConsumer
{
    protected override async Task ConsumeAsync(
        List<MessageContext<OrderPlaced>> messages)
    {
        foreach (var ctx in messages)
        {
            try
            {
                await Fulfil(ctx.Message);
                await AcknowledgeAsync(ctx);   // handled → delete from the queue
            }
            catch (Exception ex)
            {
                ex.PrintInConsole(tag: nameof(ConsumeAsync));
                // Leave it un-acked → SQS redelivers, then DLQs after N tries.
                Response.BatchItemFailures.Add(new SQSBatchResponse.BatchItemFailure
                    { ItemIdentifier = ctx.MessageId });
            }
        }
    }
}
```

## Register

```csharp
// Global SQS config, then register each consumer.
builder.Services
    .AddAwsSqsGlobalQueueConfiguration(options =>
    {
        options.AWSRegion         = appSettings.AwsRegion;
        options.Namespace         = appSettings.Stage;   // per-stage queue isolation
        options.UseDeadLetterQueue = true;               // sibling DLQ auto-created
    })
    .AddQueueConsumer<OrderPlacedConsumer>(options =>
    {
        options.QueueName                 = "order-events";
        options.MaximumReceiveMessageCount = 5;
        options.MaximumMessageRetry        = 5;          // then → DLQ
    });
```

One global SQS configuration, then a fluent `AddQueueConsumer<T>` per consumer. `UseDeadLetterQueue` auto-provisions a sibling DLQ; `MaximumMessageRetry` sets how many redeliveries a message gets before it lands there.

## Dispatch from Lambda

In an SQS-triggered function, hand the event batch to the consumer and return the partial-batch response:

```csharp
// SQS-triggered Lambda: dispatch the event batch to the consumer and return
// the partial-batch response SQS expects.
public Task<SQSBatchResponse> Handle(SQSEvent sqsEvent)
    => ServiceProvider.ProcessSqsEventWithQueueConsumerAsync<OrderPlacedConsumer>(sqsEvent);
```

## What ships with it

- `IQueueManager.EnqueueAsync<T>(message)` — typed producer, JSON serialized
- `[QueueMessage("queue")]` on the message; `[QueueConsumer(QueueName, QueueNamespace, Message)]` on the consumer
- Generated `AbstractQueueConsumer<T>` with a batched `ConsumeAsync` + `AcknowledgeAsync` — no receive-loop boilerplate
- Partial-batch failures via `Response.BatchItemFailures` — only un-acked messages redeliver
- Automatic DLQ provisioning (`UseDeadLetterQueue`) + retry budget (`MaximumMessageRetry`)
- `ProcessSqsEventWithQueueConsumerAsync<T>` for the Lambda entry point; `Namespace` for per-stage isolation

> **Free under the Community licence:** Along with `Kanject.Core.Api`, the Queue provider is free for individuals, open source, education, and companies under $250K — and published on public NuGet.

---
_Source: https://www.kanject.com/docs/core-queue/ · Kanject Docs_