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:

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:

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:

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:

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:

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):

apiVersion: kmsvc.io/v1
kind: Queue
metadata:
  name: checkout-fifo
spec:
  fifoQueue: true
  visibilityTimeoutSeconds: 60
  partitionsPerShard: 1

Dead-letter queue (failed messages):

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:

# 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:

# Get token (device code flow)
core secrets login

# Use token
export JWT_TOKEN=$(core 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:

// 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:

# 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.