homelab/project-usage/sqs-messaging.md

# SQS-like Message Queue Service (kmsvc)

**Endpoint:** `https://kmsvc.riotpiao.homelab.com` (REST + gRPC-Gateway)  
**Internal:** `kmsvc-management-service.sqs.svc.cluster.local:8080`  
**Namespace:** `sqs`

## When to Use

- **Decouple services** — Producer doesn't wait for consumer
- **Async jobs** — Fire-and-forget processing (batch, email, webhooks)
- **FIFO ordering** — Guarantee message order within `MessageGroupId`
- **Durable delivery** — At-least-once (messages in Kafka, replicated 3×)

## Quick Start

**1. Create a queue:**
```bash
kubectl apply -f - <<EOF
apiVersion: kmsvc.io/v1
kind: Queue
metadata:
  name: orders
spec:
  fifoQueue: false                           # standard queue
  visibilityTimeoutSeconds: 30               # re-deliver if not acked
  messageRetentionPeriodSeconds: 345600      # 4 days
  partitionsPerShard: 6
  maxReceiveCount: 5                         # move to DLQ after 5 fails
EOF
```

**2. Send message:**
```bash
curl -X POST https://kmsvc.riotpiao.homelab.com/v1/queues/orders/messages \
  -H "Authorization: Bearer $JWT_TOKEN" \
  -d '{
    "body": "{\"order_id\":123,\"total\":99.99}",
    "attributes": {"source":"web","priority":"high"}
  }'
```

**3. Receive message:**
```bash
curl "https://kmsvc.riotpiao.homelab.com/v1/queues/orders/messages?max_number_of_messages=10&wait_time_seconds=20" \
  -H "Authorization: Bearer $JWT_TOKEN"

# Response:
# {
#   "messages": [
#     {
#       "message_id": "abc-123",
#       "receipt_handle": "...",
#       "body": "{...}",
#       "attributes": {...},
#       "receive_count": 1
#     }
#   ]
# }
```

**4. Acknowledge (delete) message:**
```bash
curl -X DELETE "https://kmsvc.riotpiao.homelab.com/v1/queues/orders/messages/$receipt_handle" \
  -H "Authorization: Bearer $JWT_TOKEN"
```

## Configuration

| Key | Value |
|-----|-------|
| Kafka bootstrap | `kmsvc-kafka-bootstrap.sqs.svc.cluster.local:9092` |
| Redis | `kmsvc-redis-master.sqs.svc.cluster.local:6379` |
| Topic naming | `kmsvc.{queueName}.shard-{id}` |
| Replication | 3 replicas, min.insync.replicas=2 |
| Retention | 4 days (configurable per queue) |

## Common Patterns

**Batch processing:**
```bash
for i in {1..100}; do
  curl -X POST https://kmsvc.riotpiao.homelab.com/v1/queues/jobs/messages \
    -H "Authorization: Bearer $JWT_TOKEN" \
    -d "{\"body\":\"task-$i\"}" &
done
wait
```

**FIFO queue (order guaranteed per group):**
```yaml
apiVersion: kmsvc.io/v1
kind: Queue
metadata:
  name: checkout-fifo
spec:
  fifoQueue: true
  visibilityTimeoutSeconds: 60
  partitionsPerShard: 1
```

**Dead-letter queue (failed messages):**
```yaml
apiVersion: kmsvc.io/v1
kind: Queue
metadata:
  name: orders-dlq
spec:
  fifoQueue: false

---
apiVersion: kmsvc.io/v1
kind: Queue
metadata:
  name: orders
spec:
  fifoQueue: false
  maxReceiveCount: 3
  deadLetterTargetQueue: orders-dlq  # auto-route failures here
```

## Monitoring

**Grafana dashboard:** `svc-kmsvc` (automatically loaded)

**Key metrics:**
- `kmsvc_messages_sent_total` — total sent
- `kmsvc_messages_received_total` — total received
- `kmsvc_queue_depth` — pending messages per queue
- `kmsvc_message_visibility_timeout_seconds` — visibility window

**Redis in-flight tracking:**
```bash
# Connect to Redis
k port-forward -n sqs svc/redis 6379:6379 &
redis-cli

# Check pending messages
KEYS "kmsvc:pending:orders:*"
KEYS "kmsvc:inflight:*" | wc -l
```

## Authentication

**Requires JWT from Authentik:**
```bash
# Get token (device code flow)
talos secrets login

# Use token
export JWT_TOKEN=$(talos get cluster/kmsvc/jwt-token --key jwt-token)
curl -H "Authorization: Bearer $JWT_TOKEN" https://kmsvc.riotpiao.homelab.com/v1/queues
```

## Integration Example

**Story Crater backend consumer:**
```go
// Receive messages
messages, err := kmsvc.ReceiveMessage(ctx, &kmsvc.ReceiveMessageRequest{
  QueueName:          "story-crater",
  MaxNumberOfMessages: 10,
  WaitTimeSeconds:    20,
})

// Process
for _, msg := range messages.Messages {
  processMessage(msg.Body)
  
  // Acknowledge on success
  kmsvc.DeleteMessage(ctx, &kmsvc.DeleteMessageRequest{
    QueueName:     "story-crater",
    ReceiptHandle: msg.ReceiptHandle,
  })
}
```

## Troubleshooting

**Queue stuck / high lag:**
```bash
# Check Kafka broker status
k exec -n sqs pod/kmsvc-kafka-0 -- kafka-broker-api-versions.sh --bootstrap-server localhost:9092

# Inspect queue topics
k exec -n sqs pod/kmsvc-kafka-0 -- kafka-topics.sh --bootstrap-server localhost:9092 --list | grep orders
```

**Messages not being consumed:**
- Check `maxReceiveCount` (may be routing to DLQ)
- Verify consumer has `ReceiveMessage` permission (JWT scope)
- Check Redis: `KEYS "kmsvc:fifo_lock:orders:*"` (may be blocked by visibility timeout)

See `/TROUBLESHOOTING.md` for full incident guide.